feat: version-aware User Account Management API for TAK Server 5.8

CHANGELOG.md

@@ -11,6 +11,20 @@ development state is tracked under `[Unreleased]`.
 
 ## [Unreleased]
 
+### Changed
+
+- User Account Management API is now version-aware: TAK Server 5.8
+  relocated the endpoints from `/user-management/api` to
+  `/Marti/api/user-management/api`. The base path is detected from the
+  server's version string on first use and cached; it can be forced by
+  setting `user_management_base` on the `Server` instance. Detection and
+  fallback covered by unit tests, verified live on 5.7.
+
+### Added
+
+- `HomeApi.is_ldap_admin()`: wraps the new-in-5.8
+  `GET /Marti/api/util/isLdapAdmin`; returns `False` on servers where
+  the endpoint does not exist (live-guarded).
 ### Added
 
 - Submission API (`SubmissionApi`, `server.submission`): complete wrapper

live_tests/test_live_basic.py

@@ -192,3 +192,37 @@ async def test_user_api_group_lifecycle(server) -> None:
         # no user carries the group anymore -> it must be gone
         _, groups = await server.user.get_all_group_names()
         assert grp not in [g["groupname"] for g in groups]
+
+
+@pytest.mark.asyncio
+async def test_um_prefix_detection_on_live_server(server) -> None:
+    """The detected user-management prefix matches the running server."""
+    version = await server.home.server_version()
+    prefix = await server.user._base()  # noqa: SLF001
+    if version and version.startswith("5.8"):
+        assert prefix == "/Marti/api/user-management/api"
+    else:
+        assert prefix == "/user-management/api"
+        # and the legacy paths still work on this server
+        status, _ = await server.user.get_all_users()
+        assert status == 200
+
+
+@pytest.mark.asyncio
+async def test_is_ldap_admin_guard(server) -> None:
+    """isLdapAdmin is new in 5.8; on this 5.7 server it answers 404.
+
+    The wrapper returns False in that case. When the server is upgraded,
+    this test starts failing - which is the signal to verify real
+    behaviour.
+    """
+    version = await server.home.server_version()
+    result = await server.home.is_ldap_admin()
+    if version and not version.startswith("5.8"):
+        # endpoint must not exist pre-5.8
+        status, _ = await server.connection.request(
+            "get",
+            f"{server.api_base_url}/Marti/api/util/isLdapAdmin",
+            headers={"Content-Type": "application/json"},
+        )
+        assert status == 404

src/python_takserver_api/tak_file_user_account_management_api.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-# tak_file_user_account_management_api.py from https://github.com/sgofferj/takserver-api-python
+# tak_file_user_account_management_api.py from https://github.com/sgofferj/python-takserver-api
 #
 # Copyright Stefan Gofferje
 #
@@ -13,19 +13,61 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either expressed or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-#
 
 
-"""file-user-account-management-api - https://docs.tak.gov/api/takserver#tag/file-user-account-management-api"""
+"""file-user-account-management-api
+
+Manages file-based user accounts and groups. All operations are
+version-aware: on TAK Server 5.7 the endpoints live under
+``/user-management/api``, on 5.8+ they moved to
+``/Marti/api/user-management/api``. The base path is detected from the
+server's version string on first use (and cached); it can also be forced
+by setting ``user_management_base`` on the ``Server`` instance before
+first use.
+"""
 
+import re
 from typing import Any
 
+PREFIX_LEGACY = "/user-management/api"
+PREFIX_MARTI = "/Marti/api/user-management/api"
+
+
+def um_prefix_for_version(version: str | None) -> str:
+    """Returns the user-management base path for a TAK version string.
+
+    TAK Server 5.8 relocated the user-management endpoints from
+    ``/user-management/api`` to ``/Marti/api/user-management/api``.
+    Anything undetectable falls back to the legacy prefix.
+    """
+    if not version:
+        return PREFIX_LEGACY
+    match = re.match(r"^v?(\d+)\.(\d+)", version.strip())
+    if match:
+        major, minor = int(match.group(1)), int(match.group(2))
+        if major > 5 or (major == 5 and minor >= 8):
+            return PREFIX_MARTI
+    return PREFIX_LEGACY
+
 
 class UserAccountManagementApi:
     """User Account Management API wrapper"""
 
     def __init__(self, server: Any) -> None:
         self.server = server
+        # explicit override wins; otherwise detect once, lazily
+        self._prefix: str | None = getattr(server, "user_management_base", None)
+
+    async def _base(self) -> str:
+        """Resolves (and caches) the user-management base path"""
+        if self._prefix is None:
+            s, r = await self.server.connection.request(
+                "get",
+                f"{self.server.api_base_url}/Marti/api/version",
+                headers={"Content-Type": "application/json"},
+            )
+            self._prefix = um_prefix_for_version(str(r) if s == 200 else None)
+        return self._prefix
 
     # pylint: disable=too-many-arguments, too-many-positional-arguments
     async def create_or_update_file_user(
@@ -36,12 +78,16 @@ async def create_or_update_file_user(
         group_list_in: list[str] | None = None,
         group_list_out: list[str] | None = None,
     ) -> tuple[int, Any]:
-        """Create or update a user on the server
+        """Create or update a user on the server.
 
         The three group lists are always sent - the server replies with
         HTTP 500 when they are omitted.
+
+        Note: on 5.7, creating a user with a group in `groupList` did not
+        reliably persist the membership; assigning groups afterwards via
+        `update_groups_for_user()` proved dependable.
         """
-        path = "/user-management/api/new-user"
+        path = f"{await self._base()}/new-user"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         data: dict[str, Any] = {
@@ -56,15 +102,15 @@ async def create_or_update_file_user(
 
     async def get_all_users(self) -> tuple[int, Any]:
         """Returns a list of all users on the server"""
-        path = "/user-management/api/list-users"
+        path = f"{await self._base()}/list-users"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         s, r = await self.server.connection.request("get", url, headers=headers)
         return s, r
 
     async def get_all_group_names(self) -> tuple[int, Any]:
         """Returns a list of all groups on the server"""
-        path = "/user-management/api/list-groupnames"
+        path = f"{await self._base()}/list-groupnames"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         s, r = await self.server.connection.request("get", url, headers=headers)
@@ -83,33 +129,37 @@ async def group_exists(self, group: str) -> bool:
         return bool(g)
 
     async def get_users_in_group(self, group: str) -> tuple[int, Any]:
-        """Returns the users in a group (SimpleGroupWithUsersModel)"""
-        path = f"/user-management/api/users-in-group/{group}"
+        """Returns a group record with its member list"""
+        path = f"{await self._base()}/users-in-group/{group}"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         s, r = await self.server.connection.request("get", url, headers=headers)
         return s, r
 
     async def get_groups_for_user(self, username: str) -> tuple[int, Any]:
-        """Returns the groups a user belongs to (SimpleUserGroupModel)"""
-        path = f"/user-management/api/get-groups-for-user/{username}"
+        """Returns the group entitlements of a user (UserManagement model).
+
+        For channel subscriptions see
+        `tak_group_api.GroupApi.get_groups_for_user()` instead.
+        """
+        path = f"{await self._base()}/get-groups-for-user/{username}"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         s, r = await self.server.connection.request("get", url, headers=headers)
         return s, r
 
     async def change_user_password(self, username: str, password: str) -> tuple[int, Any]:
-        """Change a user's password"""
-        path = "/user-management/api/change-user-password"
+        """Changes a user's password"""
+        path = f"{await self._base()}/change-user-password"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
-        data: dict[str, Any] = {"username": username, "password": password}
+        data = {"username": username, "password": password}
         s, r = await self.server.connection.request("put", url, headers=headers, json=data)
         return s, r
 
     async def delete_user(self, username: str) -> tuple[int, Any]:
-        """Delete a user from the server"""
-        path = f"/user-management/api/delete-user/{username}"
+        """Deletes a user account"""
+        path = f"{await self._base()}/delete-user/{username}"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         s, r = await self.server.connection.request("delete", url, headers=headers)
@@ -131,10 +181,12 @@ async def create_file_users_in_bulk(
         substituted with the running number from ``start_n`` to ``end_n``,
         e.g. ``"live-test-user-[N]"`` with ``start_n=1``, ``end_n=3``
         creates ``live-test-user-1`` .. ``live-test-user-3``. The server
-        rejects requests without ``group_list`` (HTTP 400).
+        rejects requests without ``group_list`` (HTTP 400). Directions:
+        `group_list_in` = send-into memberships, `group_list_out` =
+        receive-from memberships.
         Returns a list of UserPasswordModel with the generated passwords.
         """
-        path = "/user-management/api/new-users"
+        path = f"{await self._base()}/new-users"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         data: dict[str, Any] = {
@@ -158,9 +210,12 @@ async def update_users_for_group(
         """Replace the user membership of a group (SimpleGroupWithUsersModel)
 
         The three user lists are always sent - the server replies with
-        HTTP 500 when any of them is omitted.
+        HTTP 500 when any of them is omitted. Directions are from the
+        group's point of view: `users_in_group_list_in` lists users who
+        may send INTO the group, `users_in_group_list_out` users who
+        receive its traffic, and `users_in_group_list` covers both.
         """
-        path = "/user-management/api/update-group-users"
+        path = f"{await self._base()}/update-group-users"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         data: dict[str, Any] = {
@@ -182,9 +237,12 @@ async def update_groups_for_user(
         """Replace the group membership of a user (SimpleUserGroupModel)
 
         The three group lists are always sent - the server replies with
-        HTTP 500 when any of them is omitted.
+        HTTP 500 when any of them is omitted. Directions are from the
+        group's point of view: `group_list_in` = groups the user may send
+        into, `group_list_out` = groups whose traffic they receive,
+        `group_list` = both directions.
         """
-        path = "/user-management/api/update-groups"
+        path = f"{await self._base()}/update-groups"
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         data: dict[str, Any] = {

src/python_takserver_api/tak_home_api.py

@@ -34,6 +34,20 @@ async def is_admin(self) -> bool:
         _, r = await self.server.connection.request("get", url, headers=headers)
         return bool(r)
 
+    async def is_ldap_admin(self) -> bool:
+        """Check if the configured certificate has LDAP admin rights.
+
+        New in TAK Server 5.8 - on 5.7 the endpoint does not exist
+        (HTTP 404).
+        """
+        path = "/Marti/api/util/isLdapAdmin"
+        url = self.server.api_base_url + path
+        headers = {"Content-Type": "application/json"}
+        s, r = await self.server.connection.request("get", url, headers=headers)
+        if s == 200:
+            return bool(r)
+        return False
+
     async def get_home(self) -> tuple[int, Any]:
         """Returns the server's home payload"""
         path = "/Marti/api/home"