support "user_claim_json_pointer" in create_role() for JWT/OIDC auth method

hvac/api/auth_methods/jwt.py

@@ -159,6 +159,7 @@ def create_role(
         token_period=None,
         token_type=None,
         path=None,
+        user_claim_json_pointer=None,
     ):
         """Register a role in the JWT method.
 
@@ -177,7 +178,9 @@ def create_role(
             Required for "jwt" roles, optional for "oidc" roles.
         :type bound_audiences: list
         :param user_claim: The claim to use to uniquely identify the user; this will be used as the name for the
-            Identity entity alias created due to a successful login. The claim value must be a string.
+            Identity entity alias created due to a successful login. The interpretation of the user claim
+            is configured with ``user_claim_json_pointer``. If set to ``True``, ``user_claim`` supports JSON pointer syntax
+            for referencing a claim. The claim value must be a string.
         :type user_claim: str | unicode
         :param clock_skew_leeway: Only applicable with "jwt" roles.
         :type clock_skew_leeway: int
@@ -240,6 +243,9 @@ def create_role(
         :type token_type: str
         :param path: The "path" the method/backend was mounted on.
         :type path: str | unicode
+        :param user_claim_json_pointer: Specifies if the ``user_claim`` value uses JSON pointer syntax for referencing claims.
+            By default, the ``user_claim`` value will not use JSON pointer.
+        :type user_claim_json_pointer: bool
         :return: The response of the create_role request.
         :rtype: dict
         """
@@ -269,6 +275,7 @@ def create_role(
                 "token_num_uses": token_num_uses,
                 "token_period": token_period,
                 "token_type": token_type,
+                "user_claim_json_pointer": user_claim_json_pointer,
             }
         )
         api_path = utils.format_url(

hvac/api/auth_methods/oidc.py

@@ -45,6 +45,7 @@ def create_role(
         token_period=None,
         token_type=None,
         path=None,
+        user_claim_json_pointer=None,
     ):
         """Register a role in the OIDC method.
 
@@ -63,7 +64,9 @@ def create_role(
             Required for "jwt" roles, optional for "oidc" roles.
         :type bound_audiences: list
         :param user_claim: The claim to use to uniquely identify the user; this will be used as the name for the
-            Identity entity alias created due to a successful login. The claim value must be a string.
+            Identity entity alias created due to a successful login. The interpretation of the user claim
+            is configured with ``user_claim_json_pointer``. If set to ``True``, ``user_claim`` supports JSON pointer syntax
+            for referencing a claim. The claim value must be a string.
         :type user_claim: str | unicode
         :param clock_skew_leeway: Only applicable with "jwt" roles.
         :type clock_skew_leeway: int
@@ -126,6 +129,9 @@ def create_role(
         :type token_type: str
         :param path: The "path" the method/backend was mounted on.
         :type path: str | unicode
+        :param user_claim_json_pointer: Specifies if the ``user_claim`` value uses JSON pointer syntax for referencing claims.
+            By default, the ``user_claim`` value will not use JSON pointer.
+        :type user_claim_json_pointer: bool
         :return: The response of the create_role request.
         :rtype: dict
         """
@@ -156,4 +162,5 @@ def create_role(
             token_period=token_period,
             token_type=token_type,
             path=path,
+            user_claim_json_pointer=user_claim_json_pointer,
         )

tests/unit_tests/api/auth_methods/test_jwt.py

@@ -0,0 +1,189 @@
+from hvac.api.auth_methods import JWT
+
+import requests_mock
+from parameterized import parameterized
+
+from hvac.adapters import JSONAdapter
+
+import json
+
+
+class ValueChecker:
+    def __init__(self, expected_body_params, return_status_code):
+        self.expected_body_params = expected_body_params
+        self.return_status_code = return_status_code
+        self.expected_headers = {
+            "X-Vault-Request": "true",
+            "Content-Type": "application/json",
+        }
+
+    def text_callback(self, request, context):
+        rq_params = json.loads(request.body)
+        assert rq_params == self.expected_body_params
+        rq_headers = {
+            k: v for k, v in request.headers.items() if k in self.expected_headers
+        }
+        assert rq_headers == self.expected_headers
+        context.status_code = self.return_status_code
+
+
+@parameterized.expand(
+    [
+        (
+            "hvac",
+            "https://vault/user",
+            ["https://localhost:8200/jwt-test/callback"],
+            "jwt",
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            "string",
+            False,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+        ),
+        (
+            "hvac",
+            "https://vault/user",
+            ["https://localhost:8200/jwt-test/callback"],
+            "jwt",
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            "string",
+            False,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            False,
+        ),
+        (
+            "hvac",
+            "https://vault/user",
+            ["https://localhost:8200/jwt-test/callback"],
+            "jwt",
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            "string",
+            False,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            None,
+            True,
+        ),
+    ]
+)
+@requests_mock.Mocker()
+def test_create_role(
+    name,
+    user_claim,
+    allowed_redirect_uris,
+    role_type,  # = "jwt"
+    bound_audiences,
+    clock_skew_leeway,
+    expiration_leeway,
+    not_before_leeway,
+    bound_subject,
+    bound_claims,
+    groups_claim,
+    claim_mappings,
+    oidc_scopes,
+    bound_claims_type,  # ="string"
+    verbose_oidc_logging,  # =False,
+    token_ttl,
+    token_max_ttl,
+    token_policies,
+    token_bound_cidrs,
+    token_explicit_max_ttl,
+    token_no_default_policy,
+    token_num_uses,
+    token_period,
+    token_type,
+    path,
+    user_claim_json_pointer,
+    requests_mocker,
+):
+
+    test_arguments = {
+        "name": name,
+        "role_type": role_type,
+        "bound_audiences": bound_audiences,
+        "user_claim": user_claim,
+        "clock_skew_leeway": clock_skew_leeway,
+        "expiration_leeway": expiration_leeway,
+        "not_before_leeway": not_before_leeway,
+        "bound_subject": bound_subject,
+        "bound_claims": bound_claims,
+        "groups_claim": groups_claim,
+        "claim_mappings": claim_mappings,
+        "oidc_scopes": oidc_scopes,
+        "allowed_redirect_uris": allowed_redirect_uris,
+        "bound_claims_type": bound_claims_type,
+        "verbose_oidc_logging": verbose_oidc_logging,
+        "token_ttl": token_ttl,
+        "token_max_ttl": token_max_ttl,
+        "token_policies": token_policies,
+        "token_bound_cidrs": token_bound_cidrs,
+        "token_explicit_max_ttl": token_explicit_max_ttl,
+        "token_no_default_policy": token_no_default_policy,
+        "token_num_uses": token_num_uses,
+        "token_period": token_period,
+        "token_type": token_type,
+        "user_claim_json_pointer": user_claim_json_pointer,
+    }
+
+    check_arguments = {k: v for k, v in test_arguments.items() if v is not None}
+    expected_status_code = 204
+
+    eff_path = "jwt" if path is None else path
+    mock_url = f"http://localhost:8200/v1/auth/{eff_path}/role/{name}"
+    requests_mocker.register_uri(
+        method="POST",
+        url=mock_url,
+        text=ValueChecker(check_arguments, expected_status_code).text_callback,
+    )
+    jwt = JWT(adapter=JSONAdapter())
+    actual_response = jwt.create_role(**test_arguments)
+    assert expected_status_code == actual_response.status_code