fix: harden wrappers, close review findings and add optional CA verif…

README.rst

@@ -21,6 +21,14 @@ python-takserver-api
 Async Python 3 library wrapping the TAK Server HTTP API. All network I/O uses
 ``asyncio`` + ``aiohttp`` with certificate-based mutual TLS authentication.
 
+.. note::
+   By default the client **does not verify the server certificate**
+   (``CERT_NONE``) because TAK Servers typically run self-signed certs -
+   only the client certificate is presented for mutual TLS. Pass
+   ``ca_cert="/path/to/ca.pem"`` to ``Server()`` to verify the server
+   against a CA (or pin its self-signed certificate directly); hostname
+   checking is then enabled too.
+
 Documentation lives in the project wiki:
 https://github.com/sgofferj/python-takserver-api.wiki
 

live_tests/test_live_data_feeds.py

@@ -28,6 +28,10 @@ async def feed(server):
         predicate="type == 'a-u'",
         predicate_lang="JSON_PATH",
         source_endpoint="https://localhost:8080",
+        # NEVER archive ephemeral test feeds: every CoT message would be
+        # persisted and bloat Postgres (archive=False is the helper default,
+        # pinned here explicitly so a default flip gets noticed)
+        archive=False,
         # the creating identity must be a member of a filter group,
         # otherwise the feed locks everyone out (403 Group access denied)
         filter_groups=["WOLF_Friends"],

src/python_takserver_api/class_helpers.py

@@ -22,22 +22,37 @@
 import aiohttp
 
 
+def create_client_ssl_context(certfile: str, keyfile: str, ca_file: str | None = None) -> ssl.SSLContext:
+    """Builds the mutual-TLS client context used for all server connections.
+
+    The client certificate is always presented for mTLS. When `ca_file` is
+    given, the SERVER certificate is verified against that CA (hostname
+    check included). Without it the server certificate is deliberately NOT
+    verified (`CERT_NONE`) because TAK Servers typically run with
+    self-signed certificates - only point this at servers you trust.
+    """
+    if ca_file:
+        sslcontext = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH, cafile=ca_file.strip())
+    else:
+        sslcontext = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
+        sslcontext.check_hostname = False
+        sslcontext.verify_mode = ssl.CERT_NONE
+    sslcontext.load_cert_chain(certfile=certfile, keyfile=keyfile)
+    return sslcontext
+
+
 class ConnectionHelper:
     """Helper class for managing connections"""
 
-    def __init__(self, server: Any, cert: str, key: str) -> None:
+    def __init__(self, server: Any, cert: str, key: str, ca_cert: str | None = None) -> None:
         self.server = server
         self.cert = cert.strip()
         self.key = key.strip()
+        self.ca_cert = ca_cert.strip() if ca_cert else None
 
     def get_ssl_context(self) -> aiohttp.TCPConnector:
         """Returns an SSL context for the connection"""
-        sslcontext = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
-        sslcontext.check_hostname = False
-        sslcontext.verify_mode = ssl.CERT_NONE
-        sslcontext.load_cert_chain(certfile=self.cert, keyfile=self.key)
-        tcpconn = aiohttp.TCPConnector(ssl_context=sslcontext)
-        return tcpconn
+        return aiohttp.TCPConnector(ssl_context=create_client_ssl_context(self.cert, self.key, self.ca_cert))
 
     # pylint: disable=too-many-arguments, too-many-positional-arguments
     async def request(
@@ -50,7 +65,7 @@ async def request(
     ) -> tuple[int, Any]:
         """Performs an HTTP request"""
         try:
-            r: Any = "Error"
+            r: Any
             match method:
                 case "get" | "GET":
                     # some endpoints (e.g. datafeeds/bounds/polygon) are
@@ -68,6 +83,8 @@ async def request(
                         r = await self.server.session.put(url, headers=headers, json=json)
                 case "delete" | "DELETE":
                     r = await self.server.session.delete(url, headers=headers)
+                case _:
+                    raise ValueError(f"unsupported HTTP method: {method!r}")
 
             if r.status >= 400:
                 return r.status, await r.text()

src/python_takserver_api/tak_cert_manager_api.py

@@ -120,7 +120,7 @@ async def download_certificate(self, hash_id: str) -> tuple[int, Any]:
         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, unwrap_api_response(r) if isinstance(r, dict) else r
+        return s, self._unwrap(r)
 
     async def delete_certificate(self, hash_id: str) -> tuple[int, Any]:
         """Deletes a single certificate record by its hash id.

src/python_takserver_api/tak_class.py

@@ -34,17 +34,30 @@
 class Server:
     """Takserver API helper class"""
 
-    def __init__(self, host: str, cert: str, key: str, username: str | None = None) -> None:
+    # pylint: disable=too-many-arguments, too-many-positional-arguments
+    def __init__(
+        self,
+        host: str,
+        cert: str,
+        key: str,
+        username: str | None = None,
+        ca_cert: str | None = None,
+    ) -> None:
         """Initialize a server instance.
 
         `username` is optional and only needed by APIs that address the
         authenticated user by name (e.g. the group subscription helpers).
         For certificate auth it is the certificate's CN.
+
+        `ca_cert` optionally points at a PEM file with the CA (or the
+        self-signed server certificate itself) used to VERIFY the server.
+        When given, server certificate and hostname are checked; without
+        it, verification is skipped (legacy behaviour, see README).
         """
         self.api_base_url = f"https://{host}:8443"
         self.username = username
 
-        self.connection = ConnectionHelper(self, cert, key)
+        self.connection = ConnectionHelper(self, cert, key, ca_cert)
         tcpconn = self.connection.get_ssl_context()
         self.session = aiohttp.ClientSession(connector=tcpconn)
 

src/python_takserver_api/tak_data_feed_api.py

@@ -56,7 +56,7 @@ def build_predicate_feed(
         predicate: str,
         predicate_lang: str = "JSON_PATH",
         source_endpoint: str | None = None,
-        archive: bool = True,
+        archive: bool = False,
         sync: bool = False,
         federated: bool = False,
         auth_type: str = "ANONYMOUS",
@@ -69,6 +69,10 @@ def build_predicate_feed(
         complete object. This helper fills in the fields the server needs
         and defaults `filter_groups` to `["__ANON__"]` so the feed stays
         accessible after creation (see the access-trap note above).
+
+        `archive` defaults to `False` ON PURPOSE: archived feeds persist
+        every CoT message in Postgres and quietly bloat the database.
+        Only enable archiving when you really intend to keep the data.
         """
         return {
             "name": name,

src/python_takserver_api/tak_file_user_account_management_api.py

@@ -117,16 +117,26 @@ async def get_all_group_names(self) -> tuple[int, Any]:
         return s, r
 
     async def user_exists(self, user: str) -> bool:
-        """Check if a user exists on the server"""
-        _, r = await self.get_all_users()
-        g = next((item for item in r if item["username"] == user), None)
-        return bool(g)
+        """Check if a user exists on the server.
+
+        Returns `False` when the request fails (non-200 or unparsable
+        body), not an exception - check the underlying `get_all_users()`
+        yourself if you need to distinguish "absent" from "server error".
+        """
+        s, r = await self.get_all_users()
+        if s != 200 or not isinstance(r, list):
+            return False
+        return any(isinstance(item, dict) and item.get("username") == user for item in r)
 
     async def group_exists(self, group: str) -> bool:
-        """Check if a group exists on the server"""
-        _, r = await self.get_all_group_names()
-        g = next((item for item in r if item["groupname"] == group), None)
-        return bool(g)
+        """Check if a group exists on the server.
+
+        As with `user_exists()`, a failing request yields `False`.
+        """
+        s, r = await self.get_all_group_names()
+        if s != 200 or not isinstance(r, list):
+            return False
+        return any(isinstance(item, dict) and item.get("groupname") == group for item in r)
 
     async def get_users_in_group(self, group: str) -> tuple[int, Any]:
         """Returns a group record with its member list"""

src/python_takserver_api/tak_group_api.py

@@ -199,6 +199,17 @@ def _resolve_username(self, username: str | None) -> str:
             raise ValueError("username required: pass it explicitly or construct Server(..., username=...)")
         return resolved
 
+    @staticmethod
+    def _require_group_list(status: int, payload: Any, source: str) -> list[dict[str, Any]]:
+        """Validates that a group readback succeeded and returned a list.
+
+        Raises `ValueError` with a clear message instead of letting a
+        string error body blow up later with a confusing TypeError.
+        """
+        if status != 200 or not isinstance(payload, list):
+            raise ValueError(f"{source} failed (HTTP {status}): {str(payload)[:200]}")
+        return payload
+
     async def get_active_groups(self, username: str | None = None) -> list[dict[str, Any]]:
         """Returns a user's ACTIVE CHANNEL SUBSCRIPTIONS as `{"name", "direction"}` dicts.
 
@@ -211,10 +222,11 @@ async def get_active_groups(self, username: str | None = None) -> list[dict[str,
         with (`Server(..., username="cn-of-your-cert")`).
         """
         user = self._resolve_username(username)
-        _, r = await self.get_groups_for_user(user)
+        s, r = await self.get_groups_for_user(user)
+        entries = self._require_group_list(s, r, f"get_groups_for_user({user!r})")
         seen: set[tuple[str, str]] = set()
         active: list[dict[str, Any]] = []
-        for g in r or []:
+        for g in entries:
             if not g.get("active"):
                 continue
             key = (g["name"], g["direction"])
@@ -326,8 +338,9 @@ async def get_channels(self) -> list[dict[str, Any]]:
         name spelling is preferred. SCOPE NOTE: availability, not
         subscription state - see `get_active_groups()`.
         """
-        _, r = await self.get_all_groups()
-        ordered = sorted(r or [], key=lambda g: g.get("created") or "", reverse=True)
+        s, r = await self.get_all_groups()
+        entries = self._require_group_list(s, r, "get_all_groups")
+        ordered = sorted(entries, key=lambda g: g.get("created") or "", reverse=True)
         ordered.sort(key=lambda g: g["name"] != g["name"].strip())
         channels: dict[str, dict[str, Any]] = {}
         for g in ordered:
@@ -340,8 +353,9 @@ async def get_channels(self) -> list[dict[str, Any]]:
 
     async def channel_exists(self, name: str) -> bool:
         """Check if a channel (group) is visible/available to this user"""
-        _, r = await self.get_all_groups()
-        return self._resolve_spelling(r or [], name) is not None
+        s, r = await self.get_all_groups()
+        entries = self._require_group_list(s, r, "get_all_groups")
+        return self._resolve_spelling(entries, name) is not None
 
     async def is_subscribed(
         self,
@@ -357,10 +371,11 @@ async def is_subscribed(
         by default ANY subscribed direction counts as subscribed.
         """
         user = self._resolve_username(username)
-        _, r = await self.get_groups_for_user(user)
+        s, r = await self.get_groups_for_user(user)
+        entries = self._require_group_list(s, r, f"get_groups_for_user({user!r})")
         matches = [
             g
-            for g in r or []
+            for g in entries
             if g["name"].strip() == name.strip() and (direction is None or g["direction"] == direction.upper())
         ]
         return any(g.get("active") for g in matches)

src/python_takserver_api/tak_mission_api.py

@@ -242,13 +242,12 @@ async def create_mission(
         classification: str = "",
     ) -> tuple[int, Any]:
         """Creates a mission"""
-        path = f"/Marti/api/missions/{name}?creatorUid={creator_uid}"
-        if group:
-            path += f"&group={group}"
-        if default_role:
-            path += f"&defaultRole={default_role}"
-        if classification:
-            path += f"&classification={classification}"
+        path = f"/Marti/api/missions/{name}" + _query(
+            creatorUid=creator_uid,
+            group=group or None,
+            defaultRole=default_role or None,
+            classification=classification or None,
+        )
         url = self.server.api_base_url + path
         s, r = await self.server.connection.request("put", url)
         return s, r
@@ -265,17 +264,14 @@ async def create_mission_subscription(
         end: str = "",
     ) -> tuple[int, Any]:
         """Creates a mission subscription"""
-        path = f"/Marti/api/missions/{name}/subscription?uid={uid}"
-        if topic:
-            path += f"&topic={topic}"
-        if password:
-            path += f"&password={password}"
-        if secago:
-            path += f"&secago={secago}"
-        if start:
-            path += f"&start={start}"
-        if end:
-            path += f"&end={end}"
+        path = f"/Marti/api/missions/{name}/subscription" + _query(
+            uid=uid,
+            topic=topic or None,
+            password=password or None,
+            secago=secago or None,
+            start=start or None,
+            end=end or None,
+        )
         url = self.server.api_base_url + path
         headers = {"Content-Type": "application/json"}
         s, r = await self.server.connection.request("put", url, headers=headers)
@@ -286,15 +282,15 @@ async def set_mission_role(
         self, name: str, client_uid: str, username: str, role: str, token: str
     ) -> tuple[int, Any]:
         """Sets the role for a subscriber"""
-        path = f"/Marti/api/missions/{name}/role?clientUid={client_uid}&username={username}&role={role}"
+        path = f"/Marti/api/missions/{name}/role" + _query(clientUid=client_uid, username=username, role=role)
         url = self.server.api_base_url + path
         headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
         s, r = await self.server.connection.request("put", url, headers=headers)
         return s, r
 
     async def add_mission_content(self, name: str, uids: list[str], my_uid: str, token: str) -> tuple[int, Any]:
         """Adds content to a mission"""
-        path = f"/Marti/api/missions/{name}/contents?creatorUid={my_uid}"
+        path = f"/Marti/api/missions/{name}/contents" + _query(creatorUid=my_uid)
         url = self.server.api_base_url + path
         headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
         data = {"uids": uids}
@@ -303,7 +299,7 @@ async def add_mission_content(self, name: str, uids: list[str], my_uid: str, tok
 
     async def remove_mission_content(self, name: str, uid: str, my_uid: str, token: str) -> tuple[int, Any]:
         """Removes content from a mission"""
-        path = f"/Marti/api/missions/{name}/contents?creatorUid={my_uid}&uid={uid}"
+        path = f"/Marti/api/missions/{name}/contents" + _query(creatorUid=my_uid, uid=uid)
         headers = {"Authorization": f"Bearer {token}"}
         url = self.server.api_base_url + path
         s, r = await self.server.connection.request("delete", url, headers=headers)
@@ -320,7 +316,7 @@ async def add_mission_package(
 
         The mission_package bytes can be created with build_mission_package().
         """
-        path = f"/Marti/api/missions/{name}/contents/missionpackage?creatorUid={creator_uid}"
+        path = f"/Marti/api/missions/{name}/contents/missionpackage" + _query(creatorUid=creator_uid)
         url = self.server.api_base_url + path
         headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/octet-stream"}
         s, r = await self.server.connection.request("put", url, headers=headers, data=mission_package)

src/python_takserver_api/tak_submission_api.py

@@ -27,8 +27,6 @@
 > "Invalid input name".
 >
 > **SERVER QUIRKS (verified live on 5.7-RELEASE-43-HEAD, 2026-08-25):**
-> - Input/feed names must match `[A-Za-z0-9_]` plus whitespace, max 30
->   chars - hyphens are rejected ("Invalid input name").
 > - `modify_input()` is rejected (HTTP 400) by the messaging layer for
 >   freshly created inputs.
 > - `create_data_feed()` / `modify_data_feed()` fail with an empty HTTP