fix potential vulnerability when fetching remote json data

activities/models/post.py

@@ -30,6 +30,7 @@
 )
 from core.exceptions import ActivityPubFormatError
 from core.html import ContentRenderer, FediverseHtmlParser
+from core.json import json_from_response
 from core.ld import (
     canonicalise,
     format_ld_date,
@@ -1033,8 +1034,9 @@ def by_object_uri(cls, object_uri, fetch=False) -> "Post":
                         {response.content},
                     )
                 try:
+                    json_data = json_from_response(response)
                     post = cls.by_ap(
-                        canonicalise(response.json(), include_security=True),
+                        canonicalise(json_data, include_security=True),
                         create=True,
                         update=True,
                         fetch_author=True,

activities/services/search.py

@@ -83,11 +83,11 @@ def search_url(self) -> Post | Identity | None:
         if response.status_code >= 400:
             return None
 
-        json_data = json_from_response(response)
-        if not json_data:
+        try:
+            json_data = json_from_response(response)
+            document = canonicalise(json_data, include_security=True)
+        except ValueError:
             return None
-
-        document = canonicalise(json_data, include_security=True)
         type = document.get("type", "unknown").lower()
 
         # Is it an identity?

activities/views/debug.py

@@ -5,6 +5,7 @@
 from django.utils.decorators import method_decorator
 from django.views.generic import FormView, TemplateView
 
+from core.json import json_from_response
 from core.ld import canonicalise
 from users.decorators import admin_required
 from users.models import SystemActor
@@ -50,8 +51,9 @@ def form_valid(self, form):
                 result = f"Error response: {response.status_code}\n{response.content}"
             else:
                 try:
-                    document = canonicalise(response.json(), include_security=True)
-                except json.JSONDecodeError as ex:
+                    json_data = json_from_response(response)
+                    document = canonicalise(json_data, include_security=True)
+                except ValueError as ex:
                     result = str(ex)
                 else:
                     context["raw_result"] = json.dumps(response.json(), indent=2)

core/json.py

@@ -3,19 +3,18 @@
 from httpx import Response
 
 JSON_CONTENT_TYPES = [
-    "application/json",
     "application/ld+json",
     "application/activity+json",
 ]
 
 
-def json_from_response(response: Response) -> dict | None:
+def json_from_response(response: Response) -> dict:
     content_type, *parameters = (
         response.headers.get("Content-Type", "invalid").lower().split(";")
     )
 
     if content_type not in JSON_CONTENT_TYPES:
-        return None
+        raise ValueError(f"Invalid content type: {content_type}")
 
     charset = None
 

tests/activities/models/test_post.py

@@ -13,6 +13,7 @@ def test_fetch_post(httpx_mock: HTTPXMock, config_system):
     """
     httpx_mock.add_response(
         url="https://example.com/test-actor",
+        headers={"Content-Type": "application/activity+json"},
         json={
             "@context": [
                 "https://www.w3.org/ns/activitystreams",
@@ -23,6 +24,7 @@ def test_fetch_post(httpx_mock: HTTPXMock, config_system):
     )
     httpx_mock.add_response(
         url="https://example.com/test-post",
+        headers={"Content-Type": "application/activity+json"},
         json={
             "@context": [
                 "https://www.w3.org/ns/activitystreams",

tests/api/test_search.py

@@ -86,8 +86,7 @@ def test_search_not_found(httpx_mock: HTTPXMock, api_client):
 @pytest.mark.parametrize(
     "content_type",
     [
-        "application/json",
-        "application/ld+json",
+        'application/ld+json; profile="https://www.w3.org/ns/activitystreams"',
         "application/activity+json",
     ],
 )

tests/users/models/test_identity.py

@@ -109,6 +109,7 @@ def test_fetch_actor(httpx_mock, config_system):
     # Trigger actor fetch
     httpx_mock.add_response(
         url="https://example.com/.well-known/webfinger?resource=acct:[email protected]",
+        headers={"Content-Type": "application/activity+json"},
         json={
             "subject": "acct:[email protected]",
             "aliases": [
@@ -130,6 +131,7 @@ def test_fetch_actor(httpx_mock, config_system):
     )
     httpx_mock.add_response(
         url="https://example.com/test-actor/",
+        headers={"Content-Type": "application/activity+json"},
         json={
             "@context": [
                 "https://www.w3.org/ns/activitystreams",
@@ -170,6 +172,7 @@ def test_fetch_actor(httpx_mock, config_system):
     )
     httpx_mock.add_response(
         url="https://example.com/test-actor/collections/featured/",
+        headers={"Content-Type": "application/activity+json"},
         json={
             "type": "Collection",
             "totalItems": 1,

users/models/identity.py

@@ -857,7 +857,8 @@ def fetch_pinned_post_uris(cls, uri: str) -> list[str]:
                 return []
 
         try:
-            data = canonicalise(response.json(), include_security=True)
+            json_data = json_from_response(response)
+            data = canonicalise(json_data, include_security=True)
             items: list[dict | str] = []
             if "orderedItems" in data:
                 items = list(reversed(data["orderedItems"]))
@@ -917,10 +918,8 @@ def fetch_actor(self) -> bool:
                     "Client error fetching actor: %d %s", status_code, self.actor_uri
                 )
             return False
-        json_data = json_from_response(response)
-        if not json_data:
-            return False
         try:
+            json_data = json_from_response(response)
             document = canonicalise(json_data, include_security=True)
         except ValueError:
             # servers with empty or invalid responses are inevitable