lib: use msgspec json decoder for requests Session

authentik/enterprise/endpoints/connectors/fleet/controller.py

@@ -67,7 +67,9 @@ def _paginate_hosts(self):
                     params={
                         "order_key": "hardware_serial",
                         "page": page,
-                        "per_page": 50,
+                        # Small page size as Fleet's API response, with the populate fields below
+                        # can be quite large and eat a lot of memory
+                        "per_page": 5,
                         "device_mapping": "true",
                         "populate_software": "true",
                         "populate_users": "true",

authentik/enterprise/endpoints/connectors/fleet/tests/test_connector.py

@@ -35,11 +35,11 @@ def test_sync(self):
                 text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
             )
             mock.get(
-                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=5&device_mapping=true&populate_software=true&populate_users=true",
                 json=TEST_HOST,
             )
             mock.get(
-                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=5&device_mapping=true&populate_software=true&populate_users=true",
                 json={"hosts": []},
             )
             controller.sync_endpoints()
@@ -93,11 +93,11 @@ def test_sync_headers(self):
                 text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
             )
             mock.get(
-                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=5&device_mapping=true&populate_software=true&populate_users=true",
                 json=TEST_HOST,
             )
             mock.get(
-                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=5&device_mapping=true&populate_software=true&populate_users=true",
                 json={"hosts": []},
             )
             controller.sync_endpoints()

authentik/enterprise/endpoints/connectors/fleet/tests/test_stage.py

@@ -31,11 +31,11 @@ def setUp(self):
                 text=load_fixture("fixtures/cond_acc_profile.mobileconfig"),
             )
             mock.get(
-                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=0&per_page=5&device_mapping=true&populate_software=true&populate_users=true",
                 json={"hosts": [loads(load_fixture("fixtures/host_macos.json"))]},
             )
             mock.get(
-                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=50&device_mapping=true&populate_software=true&populate_users=true",
+                "http://localhost/api/v1/fleet/hosts?order_key=hardware_serial&page=1&per_page=5&device_mapping=true&populate_software=true&populate_users=true",
                 json={"hosts": []},
             )
             controller.sync_endpoints()

authentik/lib/tests/test_utils_http.py

@@ -0,0 +1,63 @@
+"""Test http utils"""
+
+from decimal import Decimal
+
+from django.test import TestCase
+from requests.exceptions import JSONDecodeError
+
+from authentik.lib.config import CONFIG
+from authentik.lib.utils.http import (
+    DebugSession,
+    MsgspecHTTPAdapter,
+    MsgspecResponse,
+    get_http_session,
+)
+
+
+def response(content: bytes) -> MsgspecResponse:
+    """Build a response with the given body"""
+    resp = MsgspecResponse()
+    resp.status_code = 200
+    resp.encoding = "utf-8"
+    resp._content = content
+    return resp
+
+
+class TestHTTPUtils(TestCase):
+    """Test http-utils"""
+
+    def test_json(self):
+        """Test JSON decoding via msgspec"""
+        self.assertEqual(response(b'{"foo": "bar"}').json(), {"foo": "bar"})
+        self.assertEqual(response(b"[1, 2, 3]").json(), [1, 2, 3])
+
+    def test_json_invalid(self):
+        """Test invalid JSON body"""
+        with self.assertRaises(JSONDecodeError):
+            response(b"{not json").json()
+
+    def test_json_empty(self):
+        """Test empty body"""
+        with self.assertRaises(JSONDecodeError):
+            response(b"").json()
+
+    def test_json_kwargs(self):
+        """Test that json.loads kwargs fall back to the default decoder"""
+        self.assertEqual(
+            response(b'{"foo": 1.5}').json(parse_float=Decimal),
+            {"foo": Decimal("1.5")},
+        )
+
+    def test_session_adapters(self):
+        """Test that sessions from get_http_session use the msgspec adapter"""
+        session = get_http_session()
+        self.assertIsInstance(session.get_adapter("https://goauthentik.io"), MsgspecHTTPAdapter)
+        self.assertIsInstance(session.get_adapter("http://goauthentik.io"), MsgspecHTTPAdapter)
+
+    @CONFIG.patch("log_level", "trace")
+    def test_session_adapters_debug(self):
+        """Test that the debug session also uses the msgspec adapter"""
+        session = get_http_session()
+        self.assertIsInstance(session, DebugSession)
+        self.assertIsInstance(session.get_adapter("https://goauthentik.io"), MsgspecHTTPAdapter)
+        self.assertIsInstance(session.get_adapter("http://goauthentik.io"), MsgspecHTTPAdapter)

authentik/lib/utils/http.py

@@ -2,6 +2,11 @@
 
 from uuid import uuid4
 
+from msgspec import DecodeError
+from msgspec.json import Decoder
+from requests.adapters import HTTPAdapter
+from requests.exceptions import JSONDecodeError
+from requests.models import Response
 from requests.sessions import PreparedRequest, Session
 from structlog.stdlib import get_logger
 
@@ -10,13 +15,47 @@
 
 LOGGER = get_logger()
 
+_DECODER = Decoder()
+
 
 def authentik_user_agent() -> str:
     """Get a common user agent"""
     return f"authentik@{authentik_full_version()}"
 
 
-class TimeoutSession(Session):
+class MsgspecResponse(Response):
+    """requests response which decodes JSON bodies using msgspec"""
+
+    def json(self, **kwargs):
+        # msgspec's decoder doesn't support any of the options json.loads takes
+        if kwargs:
+            LOGGER.warning("Falling back to stdlib json parsing due to kwargs")
+            return super().json(**kwargs)
+        try:
+            return _DECODER.decode(self.content)
+        except DecodeError as exc:
+            raise JSONDecodeError(str(exc), self.text, 0) from exc
+
+
+class MsgspecHTTPAdapter(HTTPAdapter):
+    """HTTP adapter which returns MsgspecResponse objects"""
+
+    def build_response(self, req, resp) -> MsgspecResponse:
+        response = super().build_response(req, resp)
+        response.__class__ = MsgspecResponse
+        return response
+
+
+class BaseSession(Session):
+    """Session which decodes JSON responses using msgspec"""
+
+    def __init__(self):
+        super().__init__()
+        self.mount("https://", MsgspecHTTPAdapter())
+        self.mount("http://", MsgspecHTTPAdapter())
+
+
+class TimeoutSession(BaseSession):
     """Always set a default HTTP request timeout"""
 
     def __init__(self, default_timeout=None):