diff --git a/README.rst b/README.rst index 0ba16a5..632c8c4 100644 --- a/README.rst +++ b/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 diff --git a/live_tests/test_live_data_feeds.py b/live_tests/test_live_data_feeds.py index e8e00cf..e0feea4 100644 --- a/live_tests/test_live_data_feeds.py +++ b/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"], diff --git a/src/python_takserver_api/class_helpers.py b/src/python_takserver_api/class_helpers.py index a5491b9..73a8e1a 100644 --- a/src/python_takserver_api/class_helpers.py +++ b/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() diff --git a/src/python_takserver_api/tak_cert_manager_api.py b/src/python_takserver_api/tak_cert_manager_api.py index 4751d92..d2607fc 100644 --- a/src/python_takserver_api/tak_cert_manager_api.py +++ b/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. diff --git a/src/python_takserver_api/tak_class.py b/src/python_takserver_api/tak_class.py index 1046ab1..a3344eb 100644 --- a/src/python_takserver_api/tak_class.py +++ b/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) diff --git a/src/python_takserver_api/tak_data_feed_api.py b/src/python_takserver_api/tak_data_feed_api.py index f3a7f75..058b7e4 100644 --- a/src/python_takserver_api/tak_data_feed_api.py +++ b/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, diff --git a/src/python_takserver_api/tak_file_user_account_management_api.py b/src/python_takserver_api/tak_file_user_account_management_api.py index ea77500..77a64e5 100644 --- a/src/python_takserver_api/tak_file_user_account_management_api.py +++ b/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""" diff --git a/src/python_takserver_api/tak_group_api.py b/src/python_takserver_api/tak_group_api.py index 5199fa3..7fcdc7c 100644 --- a/src/python_takserver_api/tak_group_api.py +++ b/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) diff --git a/src/python_takserver_api/tak_mission_api.py b/src/python_takserver_api/tak_mission_api.py index e02aa2d..3e6edf5 100644 --- a/src/python_takserver_api/tak_mission_api.py +++ b/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,7 +282,7 @@ 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) @@ -294,7 +290,7 @@ async def set_mission_role( 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) diff --git a/src/python_takserver_api/tak_submission_api.py b/src/python_takserver_api/tak_submission_api.py index 00e00f6..96fa8ed 100644 --- a/src/python_takserver_api/tak_submission_api.py +++ b/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 diff --git a/tests/test_base.py b/tests/test_base.py index af4b483..1239aca 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -570,3 +570,131 @@ class MockServer: # noqa: N801 api = HomeApi(MockServer()) assert await api.server_version() is None + + +@pytest.mark.asyncio +async def test_home_api_is_ldap_admin_true() -> None: + """HomeApi.is_ldap_admin returns True on HTTP 200 with truthy data""" + from python_takserver_api.tak_home_api import HomeApi + + class MockConnection: # noqa: N801 + async def request( + self, + method: str, + url: str, + headers: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, + data: str | None = None, + ) -> tuple[int, Any]: + assert url == "https://tak.example.com:8443/Marti/api/util/isLdapAdmin" + return 200, {"ldap": True} + + class MockServer: # noqa: N801 + api_base_url: str = "https://tak.example.com:8443" + connection: Any = MockConnection() + + api = HomeApi(MockServer()) + assert await api.is_ldap_admin() is True + + +@pytest.mark.asyncio +async def test_home_api_is_ldap_admin_falsy() -> None: + """HomeApi.is_ldap_admin returns False on HTTP 200 with falsy data""" + from python_takserver_api.tak_home_api import HomeApi + + class MockConnection: # noqa: N801 + async def request( + self, + method: str, + url: str, + headers: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, + data: str | None = None, + ) -> tuple[int, Any]: + return 200, {} + + class MockServer: # noqa: N801 + api_base_url: str = "https://tak.example.com:8443" + connection: Any = MockConnection() + + api = HomeApi(MockServer()) + assert await api.is_ldap_admin() is False + + +@pytest.mark.asyncio +async def test_home_api_is_ldap_admin_missing_on_old_server() -> None: + """HomeApi.is_ldap_admin returns False when TAK < 5.8 answers 404""" + from python_takserver_api.tak_home_api import HomeApi + + class MockConnection: # noqa: N801 + async def request( + self, + method: str, + url: str, + headers: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, + data: str | None = None, + ) -> tuple[int, Any]: + return 404, "" + + class MockServer: # noqa: N801 + api_base_url: str = "https://tak.example.com:8443" + connection: Any = MockConnection() + + api = HomeApi(MockServer()) + assert await api.is_ldap_admin() is False + + +@pytest.mark.asyncio +async def test_create_mission_url_encodes_parameters() -> None: + """create_mission percent-encodes special characters in parameters""" + from python_takserver_api.tak_mission_api import MissionApi + + class MockConnection: # noqa: N801 + async def request( + self, + method: str, + url: str, + headers: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, + data: str | None = None, + ) -> tuple[int, Any]: + assert method == "put" + assert "creatorUid=T+1" in url + assert "group=a%26b%3Dc" in url + assert "&classification=" not in url.split("group=")[1] + return 200, {"name": "new"} + + class MockServer: # noqa: N801 + api_base_url: str = "https://tak.example.com:8443" + connection: Any = MockConnection() + + api = MissionApi(MockServer()) + status, _ = await api.create_mission(name="new", creator_uid="T 1", group="a&b=c") + assert status == 200 + + +@pytest.mark.asyncio +async def test_add_mission_content_url_encodes_uid() -> None: + """add_mission_content percent-encodes the creatorUid""" + from python_takserver_api.tak_mission_api import MissionApi + + class MockConnection: # noqa: N801 + async def request( + self, + method: str, + url: str, + headers: dict[str, Any] | None = None, + json: dict[str, Any] | None = None, + data: str | None = None, + ) -> tuple[int, Any]: + assert "creatorUid=TEST%2F001" in url + return 200, {"data": []} + + class MockServer: # noqa: N801 + api_base_url: str = "https://tak.example.com:8443" + connection: Any = MockConnection() + + api = MissionApi(MockServer()) + status, _ = await api.add_mission_content(name="m", uids=["u"], my_uid="TEST/001", token="t") + assert status == 200 diff --git a/tests/test_cert_manager_api.py b/tests/test_cert_manager_api.py index 934cd81..509edc3 100644 --- a/tests/test_cert_manager_api.py +++ b/tests/test_cert_manager_api.py @@ -178,3 +178,59 @@ async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> api, _ = make_api(handler) status, _ = await api.revoke_certificates(["AA", "BB"]) assert status == 200 + + +@pytest.mark.asyncio +async def test_get_active_certificates() -> None: + """get_active_certificates hits the /active endpoint and unwraps""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + assert url.endswith("/Marti/api/certadmin/cert/active") + return 200, envelope([CERT]) + + api, _ = make_api(handler) + status, certs = await api.get_active_certificates() + assert status == 200 + assert certs[0]["hash"] == "AB:CD" + + +@pytest.mark.asyncio +async def test_get_expired_certificates() -> None: + """get_expired_certificates hits the /expired endpoint""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + assert url.endswith("/Marti/api/certadmin/cert/expired") + return 200, envelope([]) + + api, _ = make_api(handler) + status, certs = await api.get_expired_certificates() + assert status == 200 + assert certs == [] + + +@pytest.mark.asyncio +async def test_get_replaced_certificates() -> None: + """get_replaced_certificates hits the /replaced endpoint""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + assert url.endswith("/Marti/api/certadmin/cert/replaced") + return 200, envelope([]) + + api, _ = make_api(handler) + status, certs = await api.get_replaced_certificates() + assert status == 200 + assert certs == [] + + +@pytest.mark.asyncio +async def test_get_revoked_certificates() -> None: + """get_revoked_certificates hits the /revoked endpoint""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + assert url.endswith("/Marti/api/certadmin/cert/revoked") + return 200, envelope([{**CERT, "revocationDate": "2026-01-01"}]) + + api, _ = make_api(handler) + status, certs = await api.get_revoked_certificates() + assert status == 200 + assert certs[0]["revocationDate"] == "2026-01-01" diff --git a/tests/test_connection.py b/tests/test_connection.py new file mode 100644 index 0000000..bfdff8f --- /dev/null +++ b/tests/test_connection.py @@ -0,0 +1,289 @@ +"""Tests for ConnectionHelper, SSL context creation and Server lifecycle""" + +import asyncio +import shutil +import ssl +import subprocess +from typing import Any + +import aiohttp +import pytest + +from python_takserver_api.class_helpers import ConnectionHelper, create_client_ssl_context +from python_takserver_api.tak_class import Server + +OPENSSL = shutil.which("openssl") + + +def _generate_keypair(tmp_path: Any, cn: str = "localhost") -> tuple[str, str]: + """Generates a throwaway self-signed keypair for SSL context tests""" + cert = tmp_path / "test.pem" + key = tmp_path / "test.key" + subprocess.run( + [ + OPENSSL, + "req", + "-x509", + "-newkey", + "rsa:2048", + "-keyout", + str(key), + "-out", + str(cert), + "-days", + "1", + "-nodes", + "-subj", + f"/CN={cn}", + ], + check=True, + capture_output=True, + ) + return str(cert), str(key) + + +def _generate_ca_signed_keypair(tmp_path: Any, cn: str = "takserver.local") -> tuple[str, str, str]: + """Generates a throwaway CA plus a server keypair signed by it. + + Returns (ca_pem, server_cert, server_key) paths. + """ + ca_cert = tmp_path / "ca.pem" + ca_key = tmp_path / "ca.key" + subprocess.run( + [ + OPENSSL, + "req", + "-x509", + "-newkey", + "rsa:2048", + "-keyout", + str(ca_key), + "-out", + str(ca_cert), + "-days", + "1", + "-nodes", + "-subj", + "/CN=test-ca", + "-addext", + "basicConstraints=critical,CA:TRUE", + ], + check=True, + capture_output=True, + ) + srv_key = tmp_path / "srv.key" + srv_csr = tmp_path / "srv.csr" + srv_cert = tmp_path / "srv.pem" + subprocess.run( + [ + OPENSSL, + "req", + "-new", + "-newkey", + "rsa:2048", + "-keyout", + str(srv_key), + "-out", + str(srv_csr), + "-nodes", + "-subj", + f"/CN={cn}", + ], + check=True, + capture_output=True, + ) + ext = tmp_path / "ext.cnf" + ext.write_text("subjectAltName=DNS:takserver.local\n") + subprocess.run( + [ + OPENSSL, + "x509", + "-req", + "-in", + str(srv_csr), + "-CA", + str(ca_cert), + "-CAkey", + str(ca_key), + "-CAcreateserial", + "-out", + str(srv_cert), + "-days", + "1", + "-extfile", + str(ext), + ], + check=True, + capture_output=True, + ) + return str(ca_cert), str(srv_cert), str(srv_key) + + +def _make_helper() -> tuple[ConnectionHelper, Any]: + class MockResponse: # noqa: N801 + status = 200 + content_type = "application/json" + + async def json(self) -> Any: + return {} + + async def text(self) -> str: + return "{}" + + class MockSession: # noqa: N801 + async def get(self, url: str, headers: Any = None, json: Any = None) -> MockResponse: + return MockResponse() + + class MockServer: # noqa: N801 + session: Any = MockSession() + + server = MockServer() + return ConnectionHelper(server, "cert.pem", "key.pem"), server + + +@pytest.mark.asyncio +async def test_request_rejects_unknown_http_method() -> None: + """request raises ValueError for unsupported methods instead of crashing""" + helper, _ = _make_helper() + with pytest.raises(ValueError, match="unsupported HTTP method"): + await helper.request("patch", "https://tak.example.com:8443/x") + + +@pytest.mark.asyncio +async def test_request_returns_json_on_success() -> None: + """request returns (status, parsed json) on a happy path""" + helper, _ = _make_helper() + status, payload = await helper.request("get", "https://tak.example.com:8443/x") + assert status == 200 + assert payload == {} + + +@pytest.mark.skipif(OPENSSL is None, reason="openssl not available") +def test_create_client_ssl_context_loads_keypair(tmp_path: Any) -> None: + """create_client_ssl_context loads the client cert and disables server verification""" + cert, key = _generate_keypair(tmp_path) + sslcontext = create_client_ssl_context(cert, key) + assert isinstance(sslcontext, ssl.SSLContext) + assert sslcontext.verify_mode == ssl.CERT_NONE # deliberate: TAK servers are self-signed + assert sslcontext.check_hostname is False + + +@pytest.mark.skipif(OPENSSL is None, reason="openssl not available") +@pytest.mark.asyncio +async def test_get_ssl_context_builds_connector_from_real_keypair(tmp_path: Any) -> None: + """get_ssl_context wraps the client SSL context in a TCPConnector""" + cert, key = _generate_keypair(tmp_path) + helper = ConnectionHelper(object(), cert, key) # noqa: B902 + connector = helper.get_ssl_context() + try: + assert isinstance(connector, aiohttp.TCPConnector) + finally: + await connector.close() + + +def _run_tls_handshake(srv_cert: str, srv_key: str, ca_file: str) -> None: + """Runs a real TLS handshake against an in-loop TLS server. + + The server presents `(srv_cert, srv_key)`; the client side uses + `create_client_ssl_context()` pinning `ca_file`. Raises `ssl.SSLError` + when the server certificate does not verify against that CA. + """ + server_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + server_ctx.load_cert_chain(certfile=srv_cert, keyfile=srv_key) + client_ctx = create_client_ssl_context(srv_cert, srv_key, ca_file=ca_file) + + async def run() -> None: + server = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0, ssl=server_ctx) + port = server.sockets[0].getsockname()[1] + try: + + async def client() -> None: + await asyncio.sleep(0.05) + _, writer = await asyncio.open_connection( + "127.0.0.1", port, ssl=client_ctx, server_hostname="takserver.local" + ) + writer.close() + + async with server: + await asyncio.wait_for(client(), timeout=5) + + finally: + server.close() + + asyncio.run(run()) + + +@pytest.mark.skipif(OPENSSL is None, reason="openssl not available") +def test_create_client_ssl_context_with_ca_verifies_server(tmp_path: Any) -> None: + """With a CA file the context enforces server certificate verification""" + ca, srv_cert, srv_key = _generate_ca_signed_keypair(tmp_path) + sslcontext = create_client_ssl_context(srv_cert, srv_key, ca_file=ca) + assert sslcontext.verify_mode == ssl.CERT_REQUIRED + assert sslcontext.check_hostname is True + + +@pytest.mark.skipif(OPENSSL is None, reason="openssl not available") +def test_handshake_succeeds_with_pinned_ca(tmp_path: Any) -> None: + """The handshake succeeds when the server CA is pinned""" + ca, srv_cert, srv_key = _generate_ca_signed_keypair(tmp_path) + _run_tls_handshake(srv_cert, srv_key, ca) + + +@pytest.mark.skipif(OPENSSL is None, reason="openssl not available") +def test_handshake_fails_with_foreign_ca(tmp_path: Any) -> None: + """The handshake fails when a foreign self-signed cert is pinned as CA""" + ca, srv_cert, srv_key = _generate_ca_signed_keypair(tmp_path) + foreign_dir = tmp_path / "foreign" + foreign_dir.mkdir() + foreign_cert, _foreign_key = _generate_keypair(foreign_dir) + with pytest.raises(ssl.SSLError): + _run_tls_handshake(srv_cert, srv_key, foreign_cert) + + +@pytest.mark.asyncio +async def test_server_close_closes_session(monkeypatch: pytest.MonkeyPatch) -> None: + """Server.close closes the underlying aiohttp session""" + + closed: list[bool] = [] + + class FakeSession: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + async def close(self) -> None: + closed.append(True) + + class FakeTCPConnector: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + monkeypatch.setattr(aiohttp, "ClientSession", FakeSession) + monkeypatch.setattr(aiohttp, "TCPConnector", FakeTCPConnector) + monkeypatch.setattr(ConnectionHelper, "get_ssl_context", lambda self: FakeTCPConnector()) + + server = Server("tak.example.com", "cert.pem", "key.pem") + assert closed == [] + await server.close() + assert closed == [True] + + +@pytest.mark.asyncio +async def test_server_passes_ca_cert_to_connection(monkeypatch: pytest.MonkeyPatch) -> None: + """Server(..., ca_cert=...) forwards the CA to the connection helper""" + + class FakeSession: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + class FakeTCPConnector: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + monkeypatch.setattr(aiohttp, "ClientSession", FakeSession) + monkeypatch.setattr(aiohttp, "TCPConnector", FakeTCPConnector) + monkeypatch.setattr(ConnectionHelper, "get_ssl_context", lambda self: FakeTCPConnector()) + + server = Server("tak.example.com", "cert.pem", "key.pem", ca_cert="ca.pem") + assert server.connection.ca_cert == "ca.pem" + server_default = Server("tak.example.com", "cert.pem", "key.pem") + assert server_default.connection.ca_cert is None diff --git a/tests/test_data_feed_api.py b/tests/test_data_feed_api.py index 431914c..5e6c91a 100644 --- a/tests/test_data_feed_api.py +++ b/tests/test_data_feed_api.py @@ -230,7 +230,11 @@ async def test_build_predicate_feed_defaults() -> None: assert feed["predicateLang"] == "JSON_PATH" assert feed["filterGroups"] == ["__ANON__"] # avoids the access lockout assert feed["authType"] == "ANONYMOUS" - assert feed["archive"] is True + assert feed["archive"] is False # safe default: archiving bloats Postgres + assert feed["sync"] is False + + feed_archived = DataFeedApi.build_predicate_feed("x", "p", archive=True) + assert feed_archived["archive"] is True feed2 = DataFeedApi.build_predicate_feed("x", "p", auth_type="X_509", filter_groups=["MY_GROUP"]) assert feed2["filterGroups"] == ["MY_GROUP"] diff --git a/tests/test_group_api.py b/tests/test_group_api.py index b4f7b55..9d4cbc8 100644 --- a/tests/test_group_api.py +++ b/tests/test_group_api.py @@ -1,5 +1,6 @@ """Tests for the Group (channel subscription) API""" +import asyncio from typing import Any import pytest @@ -380,3 +381,136 @@ async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> api, _ = make_api(handler) assert await api.is_subscribed("chan-a", username="other-user") is True + + +@pytest.mark.asyncio +async def test_unsubscribe_many_removes_all_requested_directions() -> None: + """unsubscribe_many strips the requested channels in one write""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + if url.endswith("/groups/user?username=testadmin"): + return 200, { + "version": "3", + "type": "...", + "data": [ + _group("chan-a", "IN", active=True), + _group("chan-a", "OUT", active=True), + _group("chan-b", "IN", active=True), + ], + } + assert method == "put" + assert json == [{"name": "chan-b", "direction": "IN"}] + return 200, None + + api, _ = make_api(handler) + status, _ = await api.unsubscribe_many(["chan-a"]) + assert status == 200 + + +@pytest.mark.asyncio +async def test_unsubscribe_many_keeps_other_channels() -> None: + """unsubscribe_many ignores unknown names and keeps unrelated directions""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + if url.endswith("/groups/user?username=testadmin"): + return 200, { + "version": "3", + "type": "...", + "data": [_group("chan-a", "OUT", active=True), _group("chan-b", "IN", active=True)], + } + assert json == [{"name": "chan-a", "direction": "OUT"}] + return 200, None + + api, _ = make_api(handler) + await api.unsubscribe_many(["ghost", " chan-b "], directions=["IN"]) + + +@pytest.mark.asyncio +async def test_wait_for_group_update_until_returns_on_change() -> None: + """wait_for_group_update_until returns the payload when a change arrives""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + assert "/groups/update/testadmin" in url + return 200, {"version": "3", "type": "...", "data": True} + + api, _ = make_api(handler) + status, result = await api.wait_for_group_update_until("testadmin", timeout=5.0) + assert status == 200 + assert result["data"] is True + + +@pytest.mark.asyncio +async def test_wait_for_group_update_until_times_out() -> None: + """wait_for_group_update_until raises TimeoutError when nothing changes""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + await asyncio.sleep(5) + raise AssertionError("should not complete") + + api, _ = make_api(handler) + with pytest.raises(TimeoutError): + await api.wait_for_group_update_until("testadmin", timeout=0.05) + + +@pytest.mark.asyncio +async def test_get_channels_collapses_duplicates() -> None: + """get_channels returns one record per channel with direction flags""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + assert "/groups/all" in url + return 200, { + "version": "3", + "type": "...", + "data": [ + _group("chan-b", "IN", active=True), + _group("chan-a", "IN", active=False), + _group("chan-a", "OUT", active=True), + ], + } + + api, _ = make_api(handler) + channels = await api.get_channels() + assert [c["name"] for c in channels] == ["chan-a", "chan-b"] + assert channels[0]["directions"] == {"IN": False, "OUT": True} + assert channels[1]["directions"] == {"IN": True} + + +@pytest.mark.asyncio +async def test_channel_exists_true_and_false() -> None: + """channel_exists matches padded spellings and rejects unknown names""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + return 200, { + "version": "3", + "type": "...", + "data": [_group(" chan-a ", "IN"), _group("chan-b", "OUT")], + } + + api, _ = make_api(handler) + assert await api.channel_exists("chan-a") is True + assert await api.channel_exists("chan-b") is True + assert await api.channel_exists("ghost") is False + + +@pytest.mark.asyncio +async def test_get_active_groups_raises_clear_error_on_server_error() -> None: + """get_active_groups raises ValueError instead of crashing on error text""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + return 500, "Server Error" + + api, _ = make_api(handler) + with pytest.raises(ValueError, match="HTTP 500"): + await api.get_active_groups() + + +@pytest.mark.asyncio +async def test_is_subscribed_raises_clear_error_on_server_error() -> None: + """is_subscribed raises ValueError instead of crashing on error text""" + + async def handler(method: str, url: str, headers: Any, json: Any, data: Any) -> tuple[int, Any]: + return 403, "Group access denied" + + api, _ = make_api(handler) + with pytest.raises(ValueError, match="HTTP 403"): + await api.is_subscribed("chan-a") diff --git a/tests/test_user_management.py b/tests/test_user_management.py index 28a9b3c..ddd24d6 100644 --- a/tests/test_user_management.py +++ b/tests/test_user_management.py @@ -692,3 +692,53 @@ class MockServer: # noqa: N801 api = UserAccountManagementApi(MockServer()) assert await api._base() == "/Marti/api/user-management/api" + + +@pytest.mark.asyncio +async def test_user_exists_false_on_server_error() -> None: + """user_exists returns False instead of crashing when the server errors""" + from python_takserver_api.tak_file_user_account_management_api import UserAccountManagementApi + + class MockConnection: # noqa: N801 + async def request( + self, + method: str, + url: str, + headers: dict[str, str] | None = None, + json: dict[str, Any] | None = None, + data: str | None = None, + ) -> tuple[int, Any]: + return 500, "Internal Server Error" + + class MockServer: # noqa: N801 + api_base_url: str = "https://tak.example.com:8443" + user_management_base: str = "/user-management/api" + connection: Any = MockConnection() + + api = UserAccountManagementApi(MockServer()) + assert await api.user_exists("op1") is False + + +@pytest.mark.asyncio +async def test_group_exists_false_on_server_error() -> None: + """group_exists returns False instead of crashing when the server errors""" + from python_takserver_api.tak_file_user_account_management_api import UserAccountManagementApi + + class MockConnection: # noqa: N801 + async def request( + self, + method: str, + url: str, + headers: dict[str, str] | None = None, + json: dict[str, Any] | None = None, + data: str | None = None, + ) -> tuple[int, Any]: + return 403, "Forbidden" + + class MockServer: # noqa: N801 + api_base_url: str = "https://tak.example.com:8443" + user_management_base: str = "/user-management/api" + connection: Any = MockConnection() + + api = UserAccountManagementApi(MockServer()) + assert await api.group_exists("admins") is False