Merge branch 'hvac:main' into ldap-refactor

hvac/api/auth_methods/token.py

@@ -556,6 +556,8 @@ def create_or_update_role(
         path_suffix=None,
         allowed_entity_aliases=None,
         mount_point=DEFAULT_MOUNT_POINT,
+        token_period=None,
+        token_explicit_max_ttl=None,
     ):
         """Create (or replace) the named role.
 
@@ -586,6 +588,10 @@ def create_or_update_role(
         :type path_suffix: str
         :param allowed_entity_aliases: not case sensitive.
         :type allowed_entity_aliases: str
+        :param token_period: the token will have no maximum TTL, every renewal will use the given period.
+        :type token_period: str
+        :param token_explicit_max_ttl: the token cannot be renewed past this TTL value.
+        :type token_explicit_max_ttl: str
         :param mount_point: The "path" the method/backend was mounted on.
         :type mount_point: str
         :return: The response of the create_or_update_role request.
@@ -599,6 +605,8 @@ def create_or_update_role(
                 "renewable": renewable,
                 "path_suffix": path_suffix,
                 "allowed_entity_aliases": allowed_entity_aliases,
+                "token_period": token_period,
+                "token_explicit_max_ttl": token_explicit_max_ttl,
             }
         )
         api_path = "/v1/auth/{mount_point}/roles/{role_name}".format(

hvac/api/secrets_engines/identity.py

@@ -335,7 +335,12 @@ def list_entities_by_name(self, method="LIST", mount_point=DEFAULT_MOUNT_POINT):
         return response
 
     def merge_entities(
-        self, from_entity_ids, to_entity_id, force=None, mount_point=DEFAULT_MOUNT_POINT
+        self,
+        from_entity_ids,
+        to_entity_id,
+        force=None,
+        mount_point=DEFAULT_MOUNT_POINT,
+        conflicting_alias_ids_to_keep=None,
     ):
         """Merge many entities into one entity.
 
@@ -353,6 +358,12 @@ def merge_entities(
         :type force: bool
         :param mount_point: The "path" the method/backend was mounted on.
         :type mount_point: str | unicode
+        :param conflicting_alias_ids_to_keep: A list of entity aliases to keep in the case where the to-Entity and
+            from-Entity have aliases with the same mount accessor. In the case where alias share mount accessors, the
+            alias ID given in this list will be kept or merged, and the other alias will be deleted. Note that merges
+            requiring this parameter must have only one from-Entity.
+            Requires Vault 1.12 or higher
+        :type conflicting_alias_ids_to_keep: list
         :return: The response of the request.
         :rtype: requests.Response
         """
@@ -361,6 +372,7 @@ def merge_entities(
                 "from_entity_ids": from_entity_ids,
                 "to_entity_id": to_entity_id,
                 "force": force,
+                "conflicting_alias_ids_to_keep": conflicting_alias_ids_to_keep,
             }
         )
         api_path = utils.format_url(

tests/integration_tests/api/secrets_engines/test_identity.py

@@ -610,7 +610,7 @@ def test_merge_entities(self, label, raises=None, exception_message=""):
             name="%s2" % self.TEST_ENTITY_NAME,
             mount_point=self.TEST_MOUNT_POINT,
         )
-        logging.debug("create_response2: %s" % create_response)
+        logging.debug("create_response2: %s" % create_response2)
         to_entity_id = create_response["data"]["id"]
         from_entity_ids = [create_response2["data"]["id"]]
         if raises:
@@ -636,6 +636,77 @@ def test_merge_entities(self, label, raises=None, exception_message=""):
                 second=True,
             )
 
+    @parameterized.expand(
+        [
+            param(
+                "merge success",
+            ),
+            param(
+                "merge failure",
+            ),
+        ]
+    )
+    @skipIf(
+        utils.vault_version_lt("1.12.0"),
+        '"conflicting_alias_ids_to_keep" added in Vault v1.12.0',
+    )
+    def test_merge_entities_conflicting(self, label, raises=None, exception_message=""):
+        create_response = self.client.secrets.identity.create_or_update_entity(
+            name=self.TEST_ENTITY_NAME,
+            mount_point=self.TEST_MOUNT_POINT,
+        )
+        logging.debug("create_response: %s" % create_response)
+        create_response2 = self.client.secrets.identity.create_or_update_entity(
+            name="%s2" % self.TEST_ENTITY_NAME,
+            mount_point=self.TEST_MOUNT_POINT,
+        )
+        logging.debug("create_response2: %s" % create_response2)
+        create_response3 = self.client.secrets.identity.create_or_update_entity(
+            name="%s3" % self.TEST_ENTITY_NAME,
+            mount_point=self.TEST_MOUNT_POINT,
+        )
+        logging.debug("create_response3: %s" % create_response3)
+        parent_id = create_response["data"]["id"]
+        merge_id1 = create_response2["data"]["id"]
+        merge_id2 = create_response3["data"]["id"]
+
+        merge_entities_response = self.client.secrets.identity.merge_entities(
+            from_entity_ids=[merge_id1],
+            to_entity_id=parent_id,
+            mount_point=self.TEST_MOUNT_POINT,
+        )
+        logging.debug("merge_entities_response: %s" % merge_entities_response)
+
+        if raises:
+            with self.assertRaises(raises) as cm:
+                self.client.secrets.identity.merge_entities(
+                    from_entity_ids=merge_id2,
+                    to_entity_id=parent_id,
+                    mount_point=self.TEST_MOUNT_POINT,
+                    conflicting_alias_ids_to_keep=[merge_id1],
+                )
+            self.assertIn(
+                member=exception_message,
+                container=str(cm.exception),
+            )
+        else:
+            merge_conflicting_entities_response = (
+                self.client.secrets.identity.merge_entities(
+                    from_entity_ids=merge_id2,
+                    to_entity_id=parent_id,
+                    mount_point=self.TEST_MOUNT_POINT,
+                    conflicting_alias_ids_to_keep=[merge_id1],
+                )
+            )
+            logging.debug(
+                "merge_conflicting_entities_response: %s"
+                % merge_conflicting_entities_response
+            )
+            self.assertEqual(
+                first=bool(merge_conflicting_entities_response),
+                second=True,
+            )
+
     @parameterized.expand(
         [
             param(

tests/unit_tests/api/auth_methods/test_token.py

@@ -0,0 +1,63 @@
+import pytest
+
+from unittest import mock
+from hvac.api.auth_methods.token import Token
+from hvac.adapters import Adapter
+
+
+# TODO: move this to a conftest.py somewhere
+class MockAdapter(Adapter):
+    def __init__(self, *args, **kwargs):
+        kwargs["session"] = mock.MagicMock()
+        super().__init__(*args, **kwargs)
+
+    def request(self, *args, **kwargs):
+        return (args, kwargs)
+
+    def get_login_token(self, response):
+        raise NotImplementedError()
+
+
+@pytest.fixture
+def token_auth():
+    return Token(MockAdapter())
+
+
+class TestToken:
+    @pytest.mark.parametrize("allowed_policies", ["allowed_policies", None])
+    @pytest.mark.parametrize("disallowed_policies", ["disallowed_policies", None])
+    @pytest.mark.parametrize("orphan", ["orphan", None])
+    @pytest.mark.parametrize("renewable", ["renewable", None])
+    @pytest.mark.parametrize("path_suffix", ["path_suffix", None])
+    @pytest.mark.parametrize("allowed_entity_aliases", ["allowed_entity_aliases", None])
+    @pytest.mark.parametrize("token_period", ["token_period", None])
+    @pytest.mark.parametrize("token_explicit_max_ttl", ["token_explicit_max_ttl", None])
+    def test_create_or_update_role_optional_parameters(
+        self,
+        token_auth,
+        allowed_policies,
+        disallowed_policies,
+        orphan,
+        renewable,
+        path_suffix,
+        allowed_entity_aliases,
+        token_period,
+        token_explicit_max_ttl,
+    ):
+        params = {
+            "allowed_policies": allowed_policies,
+            "disallowed_policies": disallowed_policies,
+            "orphan": orphan,
+            "renewable": renewable,
+            "path_suffix": path_suffix,
+            "allowed_entity_aliases": allowed_entity_aliases,
+            "token_period": token_period,
+            "token_explicit_max_ttl": token_explicit_max_ttl,
+        }
+        expected = params.copy()
+
+        _, rkwargs = token_auth.create_or_update_role("role_name", **params)
+
+        assert "json" in rkwargs
+        for key, value in expected.items():
+            assert value is None or rkwargs["json"][key] == value