Merge pull request #33 from Conjur-Enterprise/dry-run

CNJR-11338: Added dry-run parameter on policy methods

Author: gl-johnson

CHANGELOG.md

@@ -6,6 +6,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+## [0.1.9] - 2025-12-31
+
+### Added
+- Support dry_run parameter for policy load methods, based on [PR #51](https://github.com/cyberark/conjur-api-python/pull/51). (CNJR-11338)
+
 ## [0.1.8] - 2025-11-07
 
 ### Changed

README.md

@@ -200,22 +200,24 @@ Sets a variable to a specific value based on its ID.
 
 Note: Policy to create the variable must have already been loaded, otherwise you will get a 404 error during invocation.
 
-#### `load_policy_file(policy_name, policy_file)`
+#### `load_policy_file(policy_name, policy_file, dry_run=False)`
 
 Applies a file-based YAML to a named policy. This method only supports additive changes. Result is a dictionary object
-constructed from the returned JSON data.
+constructed from the returned JSON data. If `dry_run` is set to `True`, the policy will not be applied and the result will
+be a dictionary representing the changes that would be applied.
 
-#### `replace_policy_file(policy_name, policy_file)`
+#### `replace_policy_file(policy_name, policy_file, dry_run=False)`
 
 Replaces a named policy with one from the provided file. This is usually a destructive invocation. Result is a
-dictionary object constructed from the returned JSON data.
+dictionary object constructed from the returned JSON data. If `dry_run` is set to `True`, the policy will not be applied
+and the result will be a dictionary representing the changes that would be applied.
 
-#### `update_policy_file(policy_name, policy_file)`
+#### `update_policy_file(policy_name, policy_file, dry_run=False)`
 
 Modifies an existing Secrets Manager policy. Data may be explicitly deleted using the `!delete`, `!revoke`, and `!deny`
-statements. Unlike
-"replace" mode, no data is ever implicitly deleted. Result is a dictionary object constructed from the returned JSON
-data.
+statements. Unlike "replace" mode, no data is ever implicitly deleted. Result is a dictionary object constructed from the
+returned JSON data. If `dry_run` is set to `True`, the policy will not be applied and the result will be a dictionary
+representing the changes that would be applied.
 
 #### `list(list_constraints)`
 

ci/test/Dockerfile.test

@@ -20,14 +20,15 @@ RUN apt-get update && \
 RUN mkdir -p $INSTALL_DIR
 WORKDIR $INSTALL_DIR
 
-# Install Python 3.10.1 using pyenv, wheel and required libs
+# Install Python 3.14.x
+ENV PYTHON_VERSION=3.14
 ENV PYENV_ROOT="/root/.pyenv"
 ENV PATH="$PYENV_ROOT/bin:$PYENV_ROOT/shims:$PATH"
 
 RUN curl -L -s https://github.com/pyenv/pyenv-installer/raw/master/bin/pyenv-installer | bash \
     && eval "$(pyenv init --path)" \
-    && env PYTHON_CONFIGURE_OPTS="--enable-shared" pyenv install 3.10.1 \
-    && pyenv global 3.10.1 \
+    && env PYTHON_CONFIGURE_OPTS="--enable-shared" pyenv install $PYTHON_VERSION \
+    && pyenv global $PYTHON_VERSION \
     && pip install wheel
 
 # By putting this COPY command after we install Python, Docker will cache the Python installation layer,

conjur_api/client.py

@@ -10,8 +10,11 @@
 # Builtins
 import json
 import logging
+import re
 from typing import Optional, Tuple
 
+from packaging.version import Version
+
 from conjur_api.errors.errors import ResourceNotFoundException, MissingRequiredParameterException, HttpStatusError
 from conjur_api.http.api import Api
 from conjur_api.interface.authentication_strategy_interface import AuthenticationStrategyInterface
@@ -27,6 +30,25 @@
 
 logger = logging.getLogger(__name__)
 
+# List of possible Secrets Manager SaaS URL suffixes
+_CONJUR_CLOUD_SUFFIXES = [
+    ".cyberark.cloud",
+    ".integration-cyberark.cloud",
+    ".test-cyberark.cloud",
+    ".dev-cyberark.cloud",
+    ".cyberark-everest-integdev.cloud",
+    ".cyberark-everest-pre-prod.cloud",
+    ".sandbox-cyberark.cloud",
+    ".pt-cyberark.cloud",
+]
+
+# Build regex pattern: (\\.secretsmgr|-secretsmanager) followed by one of the suffixes
+_SUFFIXES_PATTERN = "|".join(re.escape(suffix) for suffix in _CONJUR_CLOUD_SUFFIXES)
+_CONJUR_CLOUD_REGEXP = re.compile(
+    rf"(\.secretsmgr|-secretsmanager)({_SUFFIXES_PATTERN})",
+    re.IGNORECASE
+)
+
 @allow_sync_invocation()
 # pylint: disable=too-many-public-methods
 class Client:
@@ -39,7 +61,8 @@ class Client:
     try:
         integration_version = version("conjur_api")
     except PackageNotFoundError:
-        integration_version = 'No Version Found'
+        # setuptools defaults to "0.0.dev0" (PEP 440), so we use a default version that adheres to that for testing purposes
+        integration_version = '0.0.dev0'
     integration_type = 'cybr-secretsmanager'
     vendor_name = 'CyberArk'
     vendor_version = None
@@ -95,6 +118,7 @@ def __init__(
         self.ssl_verification_mode = ssl_verification_mode
         self.connection_info = connection_info
         self.debug = debug
+        self.conjur_version = None  # Cache for server version
         self._api = self._create_api(http_debug, authn_strategy)
 
         self.set_telemetry_header()
@@ -293,23 +317,29 @@ async def set(self, variable_id: str, value: str) -> str:
         """
         await self._api.set_variable(variable_id, value)
 
-    async def load_policy_file(self, policy_name: str, policy_file: str) -> dict:
+    async def load_policy_file(self, policy_name: str, policy_file: str, dry_run: bool = False) -> dict:
         """
         Applies a file-based policy to the Conjur instance
         """
-        return await self._api.load_policy_file(policy_name, policy_file)
+        if dry_run:
+            await self._validate_dry_run_support()
+        return await self._api.load_policy_file(policy_name, policy_file, dry_run)
 
-    async def replace_policy_file(self, policy_name: str, policy_file: str) -> dict:
+    async def replace_policy_file(self, policy_name: str, policy_file: str, dry_run: bool = False) -> dict:
         """
         Replaces a file-based policy defined in the Conjur instance
         """
-        return await self._api.replace_policy_file(policy_name, policy_file)
+        if dry_run:
+            await self._validate_dry_run_support()
+        return await self._api.replace_policy_file(policy_name, policy_file, dry_run)
 
-    async def update_policy_file(self, policy_name: str, policy_file: str) -> dict:
+    async def update_policy_file(self, policy_name: str, policy_file: str, dry_run: bool = False) -> dict:
         """
         Replaces a file-based policy defined in the Conjur instance
         """
-        return await self._api.update_policy_file(policy_name, policy_file)
+        if dry_run:
+            await self._validate_dry_run_support()
+        return await self._api.update_policy_file(policy_name, policy_file, dry_run)
 
     async def rotate_other_api_key(self, resource: Resource) -> str:
         """
@@ -350,6 +380,108 @@ async def get_server_info(self):
             else:
                 raise
 
+    def _is_version_less_than(self, version1: str, version2: str) -> bool:
+        """
+        Checks if version1 is less than version2.
+        @param version1: First version string (e.g., "1.21.1", "1.21.1-beta", "1.24.0-1049")
+        @param version2: Second version string (e.g., "1.21.1")
+        @return: True if version1 < version2, False otherwise
+        """
+        return Version(version1) < Version(version2)
+
+    def _is_conjur_cloud_url(self, url: str) -> bool:
+        """
+        Checks if the URL is a Conjur Cloud (Secrets Manager SaaS) URL.
+        Matches the Go regex pattern: (\\.secretsmgr|-secretsmanager) followed by one of the cloud suffixes.
+        @param url: The Conjur URL to check
+        @return: True if the URL is a Conjur Cloud URL, False otherwise
+        """
+        if not url:
+            return False
+
+        return bool(_CONJUR_CLOUD_REGEXP.search(url))
+
+    async def server_version(self) -> str:
+        """
+        Retrieves the Conjur server version, either from the '/info' endpoint in Secrets Manager Self-Hosted,
+        or from the root endpoint in Conjur OSS. The version returned corresponds to the Conjur OSS version,
+        which in Conjur Enterprise is the version of the 'possum' service.
+        
+        The version is cached after the first retrieval to avoid making multiple requests.
+
+        @return: Server version string
+        @raises: Exception if unable to retrieve server version or if running against Conjur Cloud
+        """
+        # Return cached version if available
+        if self.conjur_version is not None:
+            return self.conjur_version
+
+        url = self.connection_info.conjur_url
+
+        if self._is_conjur_cloud_url(url):
+            raise Exception("Unable to retrieve server version: not supported in Secrets Manager SaaS")
+
+        # Try to get enterprise server info first
+        enterprise_error = None
+        try:
+            info = await self.get_server_info()
+            # Return the version of the 'possum' service, which corresponds to the Conjur OSS version
+            if isinstance(info, dict) and 'services' in info:
+                services = info.get('services', {})
+                if 'possum' in services:
+                    possum_service = services.get('possum', {})
+                    if isinstance(possum_service, dict) and 'version' in possum_service:
+                        self.conjur_version = possum_service['version']
+                        return self.conjur_version
+        except Exception as err:
+            # If enterprise info fails, try root endpoint
+            enterprise_error = err
+
+        # Try to get version from root endpoint (Conjur OSS)
+        try:
+            version = await self._api.get_server_version_from_root()
+            if version:
+                self.conjur_version = version
+                return self.conjur_version
+        except Exception as root_err:
+            # Both methods failed, raise an error with details
+            error_msg = "failed to retrieve server version"
+            if enterprise_error:
+                error_msg += f": enterprise info error - {enterprise_error}"
+            if root_err:
+                error_msg += f", root endpoint error - {root_err}"
+            raise Exception(error_msg) from root_err
+
+        raise Exception("failed to retrieve server version: both enterprise info and root endpoint failed")
+
+    async def _validate_dry_run_support(self):
+        """
+        Validates that dry_run is supported by checking:
+        1. The server is not Conjur Cloud (SaaS)
+        2. The server version is >= 1.21.1
+        
+        @raises: Exception if dry_run is not supported
+        """
+        url = self.connection_info.conjur_url
+
+        # Check if it's Conjur Cloud
+        if self._is_conjur_cloud_url(url):
+            raise Exception("dry_run is not supported in Secrets Manager SaaS")
+
+        # Check server version
+        try:
+            server_version = await self.server_version()
+            min_version = "1.21.1"
+
+            if self._is_version_less_than(server_version, min_version):
+                raise Exception(f"dry_run requires Conjur server version {min_version} or higher, but server version is {server_version}")
+        except Exception as err:
+            # If we can't get the version, we should still raise an error
+            # but include the original error message
+            error_msg = str(err)
+            raise Exception(f"Unable to validate dry_run support: {error_msg}") from err
+
+
     async def change_personal_password(
             self, logged_in_user: str, current_password: str,
             new_password: str) -> str:

conjur_api/http/api.py

@@ -7,6 +7,7 @@
 """
 # Builtins
 import logging
+import re
 from datetime import datetime
 from typing import Optional, Tuple
 from urllib import parse
@@ -488,7 +489,7 @@ async def set_variable(self, variable_id: str, value: str) -> str:
 
     async def _load_policy_file(
             self, policy_id: str, policy_file: str,
-            http_verb: HttpVerb) -> dict:
+            http_verb: HttpVerb, dry_run: bool) -> dict:
         """
         This method is used to load, replace or update a file-based policy into the desired
         name.
@@ -505,32 +506,37 @@ async def _load_policy_file(
         if api_token is None:
             raise MissingApiTokenException()
 
+        query = {}
+        if dry_run:
+            query = { 'dryRun': 'true' }
+
         response = await invoke_endpoint(http_verb, ConjurEndpoint.POLICIES, params,
                                          policy_data, api_token=api_token,
                                          ssl_verification_metadata=self.ssl_verification_data,
+                                         query=query,
                                          proxy_params=self._connection_info.proxy_params)
         return response.json
 
-    async def load_policy_file(self, policy_id: str, policy_file: str) -> dict:
+    async def load_policy_file(self, policy_id: str, policy_file: str, dry_run: bool) -> dict:
         """
         This method is used to load a file-based policy into the desired
         name.
         """
-        return await self._load_policy_file(policy_id, policy_file, HttpVerb.POST)
+        return await self._load_policy_file(policy_id, policy_file, HttpVerb.POST, dry_run)
 
-    async def replace_policy_file(self, policy_id: str, policy_file: str) -> dict:
+    async def replace_policy_file(self, policy_id: str, policy_file: str, dry_run: bool) -> dict:
         """
         This method is used to replace a file-based policy into the desired
         policy ID.
         """
-        return await self._load_policy_file(policy_id, policy_file, HttpVerb.PUT)
+        return await self._load_policy_file(policy_id, policy_file, HttpVerb.PUT, dry_run)
 
-    async def update_policy_file(self, policy_id: str, policy_file: str) -> dict:
+    async def update_policy_file(self, policy_id: str, policy_file: str, dry_run: bool) -> dict:
         """
         This method is used to update a file-based policy into the desired
         policy ID.
         """
-        return await self._load_policy_file(policy_id, policy_file, HttpVerb.PATCH)
+        return await self._load_policy_file(policy_id, policy_file, HttpVerb.PATCH, dry_run)
 
     async def rotate_other_api_key(self, resource: Resource) -> str:
         """
@@ -620,6 +626,35 @@ async def get_server_info(self):
                                      ssl_verification_metadata=self.ssl_verification_data,
                                      proxy_params=self._connection_info.proxy_params)
 
+    async def get_server_version_from_root(self) -> str:
+        """
+        Retrieves the Conjur server version from the root endpoint (Conjur OSS).
+        The root endpoint returns HTML, so we parse it to extract the version.
+        @return: Server version string
+        """
+        params = {
+            'url': self._url
+        }
+        response = await invoke_endpoint(HttpVerb.GET,
+                                         ConjurEndpoint.ROOT,
+                                         params,
+                                         ssl_verification_metadata=self.ssl_verification_data,
+                                         proxy_params=self._connection_info.proxy_params)
+        # The root endpoint returns HTML with version in format: "Version X.Y.Z" or "Version X.Y.Z-build"
+        # Extract version using regex pattern
+        html_text = response.text
+        # Pattern matches: Version followed by space and version number (e.g., "Version 1.24.0-1049")
+        version_pattern = r'Version\s+(\d+\.\d+\.\d+(?:-\d+)?)'
+        match = re.search(version_pattern, html_text, re.IGNORECASE)
+        if match:
+            return match.group(1)
+        # Fallback: try to find any semver pattern in the HTML
+        semver_pattern = r'(\d+\.\d+\.\d+(?:-\d+)?)'
+        match = re.search(semver_pattern, html_text)
+        if match:
+            return match.group(1)
+        raise Exception("Unable to extract version from root endpoint HTML response")
+
     async def whoami(self) -> dict:
         """
         This method provides dictionary of information about the user making an API request.

conjur_api/http/endpoints.py

@@ -24,6 +24,7 @@ class ConjurEndpoint(Enum):
     LOGIN = "{url}/authn/{account}/login"
     LOGIN_LDAP = "{url}/authn-ldap/{service_id}/{account}/login"
     INFO = "{url}/info"
+    ROOT = "{url}/"
     POLICIES = "{url}/policies/{account}/policy/{identifier}"
     BATCH_SECRETS = "{url}/secrets"
     SECRETS = "{url}/secrets/{account}/{kind}/{identifier}"

conjur_api/utils/decorators.py

@@ -17,18 +17,18 @@ def allow_sync_invocation():
     """
 
     def allow_sync_mode(func):
-        def wrapper(self, *args):
+        def wrapper(self, *args, **kwargs):
             should_run_async = getattr(self, "async_mode")
             should_run_async |= func.__name__.startswith("_")  # omit private functions
             if should_run_async:  # Function should remain async
-                return func(self, *args)
+                return func(self, *args, **kwargs)
             loop = _get_event_loop()
             if loop is not None and loop.is_running():
                 logger.error(
                     "Failed to run conjur_api %s function in sync mode "
                     "because code is running inside event loop", func.__name__)
                 raise SyncInvocationInsideEventLoopError()
-            return asyncio.run(func(self, *args))
+            return asyncio.run(func(self, *args, **kwargs))
 
         return wrapper