Add support for node identities to token operations

consul/api/acl/token.py

@@ -10,6 +10,19 @@
     import builtins
 
 
+def parse_identities(node_identities: list[str] | None) -> dict[str, list[dict[str, str]]]:
+    identities: list[dict[str, str]] = []
+    if node_identities:
+        for identity in node_identities:
+            try:
+                name, datacenter = identity.split(":", 1)
+            except ValueError as e:
+                raise ValueError(f"Node identity must be 'node:datacenter', got {identity!r}") from e
+            identities.append({"NodeName": name, "Datacenter": datacenter})
+        return {"NodeIdentities": identities}
+    return {}
+
+
 class AclPolicyLink(TypedDict, total=False):
     ID: str
     Name: str
@@ -100,6 +113,7 @@ def create(
         token: str | None = None,
         accessor_id: str | None = None,
         secret_id: str | None = None,
+        node_identities: builtins.list[str] | None = None,
         policies_id: builtins.list[str] | None = None,
         description: str = "",
         policies_name: builtins.list[str] | None = None,
@@ -113,6 +127,7 @@ def create(
         :param token: token with acl:write capability
         :param accessor_id: The accessor ID of the token to create
         :param secret_id: The secret ID of the token to create
+        :param node_identities: Optional list of node identities (format: 'nodename:datacenter'), requires consul>=1.8.1
         :param description: Optional new token description
         :param policies_id: Optional list of policies id
         :param roles_id: Optional list of roles id
@@ -129,6 +144,8 @@ def create(
         if description:
             json_data["Description"] = description
 
+        json_data.update(parse_identities(node_identities))
+
         policies: list[dict[str, str]] = []
         if policies_id:
             policies.extend({"ID": policy} for policy in policies_id)
@@ -165,6 +182,7 @@ def update(
         accessor_id: str,
         token: str | None = None,
         secret_id: str | None = None,
+        node_identities: builtins.list[str] | None = None,
         description: str = "",
         policies_id: builtins.list[str] | None = None,
         policies_name: builtins.list[str] | None = None,
@@ -178,6 +196,7 @@ def update(
         :param accessor_id: The accessor ID of the token to update
         :param token: token with acl:write capability
         :param secret_id: Optional secret ID of the token to update
+        :param node_identities: Optional list of node identities (format: 'nodename:datacenter'), requires consul>=1.8.1
         :param description: Optional new token description
         :param policies_id: Optional list of policies id
         :param roles_id: Optional list of roles id
@@ -187,11 +206,14 @@ def update(
         """
 
         json_data: dict[str, typing.Any] = {"AccessorID": accessor_id}
+
         if secret_id:
             json_data["SecretID"] = secret_id
         if description:
             json_data["Description"] = description
 
+        json_data.update(parse_identities(node_identities))
+
         policies: list[dict[str, str]] = []
         if policies_id:
             policies.extend({"ID": policy} for policy in policies_id)

tests/api/test_acl.py

@@ -1,3 +1,5 @@
+import typing
+
 import pytest
 
 import consul
@@ -102,11 +104,16 @@ def test_acl_token_create(self, acl_consul) -> None:
             description="some token!",
             token=master_token,
         )
+        c.acl.token.create(
+            accessor_id="00000000-0000-DEAF-0000-000000000000",
+            node_identities=["deaf:dc1"],
+            token=master_token,
+        )
 
         assert c.acl.token.read(accessor_id="00000000-DEAD-BEEF-0000-000000000000", token=master_token)
         assert c.acl.token.read(accessor_id="00000000-0000-A5A5-0000-000000000000", token=master_token)
 
-        expected = [
+        expected: list[dict[typing.Any, typing.Any]] = [
             {
                 "AccessorID": "00000000-DEAD-BEEF-0000-000000000000",
                 "Description": "",
@@ -120,6 +127,15 @@ def test_acl_token_create(self, acl_consul) -> None:
                 "SecretID": "00000000-A5A5-0000-0000-000000000000",
                 "Description": "some token!",
             },
+            {
+                "AccessorID": "00000000-0000-DEAF-0000-000000000000",
+                "NodeIdentities": [
+                    {
+                        "NodeName": "deaf",
+                        "Datacenter": "dc1",
+                    },
+                ],
+            },
         ]
         acl = c.acl.token.list(token=master_token)
         assert find_recursive(acl, expected)
@@ -155,17 +171,29 @@ def test_acl_token_update(self, acl_consul) -> None:
 
         assert len(c.acl.token.list(token=master_token)) == 2
         c.acl.token.create(
-            accessor_id="00000000-DEAD-BEEF-0000-000000000000", description="original", token=master_token
+            accessor_id="00000000-DEAD-BEEF-0000-000000000000",
+            description="original",
+            node_identities=["test01:dc1"],
+            token=master_token,
         )
         assert len(c.acl.token.list(token=master_token)) == 3
         c.acl.token.update(
-            accessor_id="00000000-DEAD-BEEF-0000-000000000000", description="updated", token=master_token
+            accessor_id="00000000-DEAD-BEEF-0000-000000000000",
+            description="updated",
+            node_identities=["test02:dc2"],
+            token=master_token,
         )
         assert len(c.acl.token.list(token=master_token)) == 3
 
         expected = {
             "AccessorID": "00000000-DEAD-BEEF-0000-000000000000",
             "Description": "updated",
+            "NodeIdentities": [
+                {
+                    "NodeName": "test02",
+                    "Datacenter": "dc2",
+                },
+            ],
         }
         acl = c.acl.token.read(accessor_id="00000000-DEAD-BEEF-0000-000000000000", token=master_token)
         assert find_recursive(acl, expected)