Adding security API (#63)

apetenchea · web-flow · commit a171df7b449c · 2025-08-08T06:39:03.000Z
diff --git a/arangoasync/database.py b/arangoasync/database.py
@@ -38,7 +38,10 @@
     PermissionListError,
     PermissionResetError,
     PermissionUpdateError,
+    ServerEncryptionError,
     ServerStatusError,
+    ServerTLSError,
+    ServerTLSReloadError,
     ServerVersionError,
     TaskCreateError,
     TaskDeleteError,
@@ -2072,6 +2075,81 @@ def response_handler(resp: Response) -> Json:
 
         return await self._executor.execute(request, response_handler)
 
+    async def tls(self) -> Result[Json]:
+        """Return TLS data (keyfile, clientCA).
+
+        This API requires authentication.
+
+        Returns:
+            dict: dict containing the following components:
+                - keyfile: Information about the key file.
+                - clientCA: Information about the Certificate Authority (CA) for client certificate verification.
+
+        Raises:
+            ServerTLSError: If the operation fails.
+
+        References:
+            - `get-the-tls-data <https://docs.arangodb.com/stable/develop/http-api/security/#get-the-tls-data>`__
+        """  # noqa: E501
+        request = Request(method=Method.GET, endpoint="/_admin/server/tls")
+
+        def response_handler(resp: Response) -> Json:
+            if not resp.is_success:
+                raise ServerTLSError(resp, request)
+            result: Json = self.deserializer.loads(resp.raw_body)["result"]
+            return result
+
+        return await self._executor.execute(request, response_handler)
+
+    async def reload_tls(self) -> Result[Json]:
+        """Reload TLS data (keyfile, clientCA).
+
+        This is a protected API and can only be executed with superuser rights.
+
+        Returns:
+            dict: New TLS data.
+
+        Raises:
+            ServerTLSReloadError: If the operation fails.
+
+        References:
+            - `reload-the-tls-data <https://docs.arangodb.com/stable/develop/http-api/security/#reload-the-tls-data>`__
+        """  # noqa: E501
+        request = Request(method=Method.POST, endpoint="/_admin/server/tls")
+
+        def response_handler(resp: Response) -> Json:
+            if not resp.is_success:
+                raise ServerTLSReloadError(resp, request)
+            result: Json = self.deserializer.loads(resp.raw_body)["result"]
+            return result
+
+        return await self._executor.execute(request, response_handler)
+
+    async def encryption(self) -> Result[Json]:
+        """Rotate the user-supplied keys for encryption.
+
+        This is a protected API and can only be executed with superuser rights.
+        This API is not available on Coordinator nodes.
+
+        Returns:
+            dict: Encryption keys.
+
+        Raises:
+            ServerEncryptionError: If the operation fails.
+
+        References:
+           - `rotate-the-encryption-keys <https://docs.arangodb.com/stable/develop/http-api/security/#rotate-the-encryption-keys>`__
+        """  # noqa: E501
+        request = Request(method=Method.POST, endpoint="/_admin/server/encryption")
+
+        def response_handler(resp: Response) -> Json:
+            if not resp.is_success:
+                raise ServerEncryptionError(resp, request)
+            result: Json = self.deserializer.loads(resp.raw_body)["result"]
+            return result
+
+        return await self._executor.execute(request, response_handler)
+
     async def list_transactions(self) -> Result[Jsons]:
         """List all currently running stream transactions.
 
diff --git a/arangoasync/exceptions.py b/arangoasync/exceptions.py
@@ -435,6 +435,10 @@ class SerializationError(ArangoClientError):
     """Failed to serialize the request."""
 
 
+class ServerEncryptionError(ArangoServerError):
+    """Failed to reload user-defined encryption keys."""
+
+
 class ServerConnectionError(ArangoServerError):
     """Failed to connect to ArangoDB server."""
 
@@ -443,6 +447,14 @@ class ServerStatusError(ArangoServerError):
     """Failed to retrieve server status."""
 
 
+class ServerTLSError(ArangoServerError):
+    """Failed to retrieve TLS data."""
+
+
+class ServerTLSReloadError(ArangoServerError):
+    """Failed to reload TLS."""
+
+
 class ServerVersionError(ArangoServerError):
     """Failed to retrieve server version."""
 
diff --git a/docs/certificates.rst b/docs/certificates.rst
@@ -108,3 +108,25 @@ Use a client certificate chain
 
 If you want to have fine-grained control over the HTTP connection, you should define
 your HTTP client as described in the :ref:`HTTP` section.
+
+Security features
+=================
+
+See the `ArangoDB Manual`_ for more information on security features.
+
+**Example:**
+
+.. code-block:: python
+
+    async with ArangoClient(hosts=url) as client:
+        db = await client.db(
+            sys_db_name, auth_method="superuser", token=token, verify=True
+        )
+
+        # Get TLS data
+        tls = await db.tls()
+
+        # Reload TLS data
+        tls = await db.reload_tls()
+
+.. _ArangoDB Manual: https://docs.arangodb.com/stable/develop/http-api/security/
diff --git a/docs/migration.rst b/docs/migration.rst
@@ -2,7 +2,7 @@ Coming from python-arango
 -------------------------
 
 Generally, migrating from `python-arango`_ should be a smooth transition. For the most part, the API is similar,
-but there are a few things to note._
+but there are a few things to note.
 
 Helpers
 =======
diff --git a/tests/test_client.py b/tests/test_client.py
@@ -3,6 +3,7 @@
 from arangoasync.auth import JwtToken
 from arangoasync.client import ArangoClient
 from arangoasync.compression import DefaultCompressionManager
+from arangoasync.exceptions import ServerEncryptionError
 from arangoasync.http import DefaultHTTPClient
 from arangoasync.resolver import DefaultHostResolver, RoundRobinHostResolver
 from arangoasync.version import __version__
@@ -131,6 +132,19 @@ async def test_client_jwt_superuser_auth(
             await db.jwt_secrets()
             await db.reload_jwt_secrets()
 
+        # Get TLS data
+        tls = await db.tls()
+        assert isinstance(tls, dict)
+
+        # Reload TLS data
+        tls = await db.reload_tls()
+        assert isinstance(tls, dict)
+
+        # Rotate
+        with pytest.raises(ServerEncryptionError):
+            # Not allowed on coordinators
+            await db.encryption()
+
     # token missing
     async with ArangoClient(hosts=url) as client:
         with pytest.raises(ValueError):
