fix(geo): reject foreign relay IDs in mutations, and don't cache private tiles

baseapp/geo/graphql/__init__.py

@@ -1,15 +0,0 @@
-from .object_types import (
-    BaseGeoJSONFeatureObjectType,
-    GeoJSONFeatureObjectType,
-    GeoJSONFeatureProperties,
-    GeometryObjectType,
-)
-from .scalars import Geometry
-
-__all__ = [
-    "BaseGeoJSONFeatureObjectType",
-    "GeoJSONFeatureObjectType",
-    "GeoJSONFeatureProperties",
-    "GeometryObjectType",
-    "Geometry",
-]

baseapp/geo/graphql/mutations.py

@@ -1,3 +1,5 @@
+import logging
+
 import graphene
 import swapper
 from django import forms
@@ -13,11 +15,41 @@
 from ..models import ALLOWED_GEOMETRY_TYPES
 from .scalars import Geometry
 
+logger = logging.getLogger(__name__)
+
 GeoJSONFeature = swapper.load_model("baseapp_geo", "GeoJSONFeature")
 app_label = GeoJSONFeature._meta.app_label
 ObjectType = GeoJSONFeature.get_graphql_object_type()
 
 
+def _resolve_relay_id(info: graphene.ResolveInfo, relay_id):
+    """Resolve a relay ID, treating any failure as "not found".
+
+    Never let the resolver's own exception reach the client: it leaks which IDs exist.
+    """
+    try:
+        return get_obj_from_relay_id(info, relay_id)
+    except Exception:
+        logger.exception("Could not resolve relay id in a geo mutation")
+        return None
+
+
+def _get_feature(info: graphene.ResolveInfo, relay_id) -> "GeoJSONFeature":
+    """Resolve a relay ID to a GeoJSONFeature, or raise `not_found`.
+
+    `get_obj_from_relay_id` resolves a public ID through DocumentId and returns the object
+    of ANY registered model, so this type check is what stops these mutations from writing
+    to (or deleting) another block's rows.
+    """
+    obj = _resolve_relay_id(info, relay_id)
+    if not isinstance(obj, GeoJSONFeature):
+        raise GraphQLError(
+            str(_("GeoJSON feature not found")),
+            extensions={"code": "not_found"},
+        )
+    return obj
+
+
 class GeoJSONFeatureForm(forms.ModelForm):
     class Meta:
         model = GeoJSONFeature
@@ -68,7 +100,12 @@ def mutate_and_get_payload(
                 extensions={"code": "permission_required"},
             )
 
-        target = get_obj_from_relay_id(info, input.get("target_object_id"))
+        target = _resolve_relay_id(info, input.get("target_object_id"))
+        if target is None:
+            raise GraphQLError(
+                str(_("Target object not found")),
+                extensions={"code": "not_found"},
+            )
         instance = GeoJSONFeature(target=target)
 
         form = GeoJSONFeatureForm(instance=instance, data=input)
@@ -121,7 +158,7 @@ def mutate_and_get_payload(
                 extensions={"code": "permission_required"},
             )
 
-        instance = get_obj_from_relay_id(info, input.get("id"))
+        instance = _get_feature(info, input.get("id"))
 
         data = {
             field: input[field] if field in input else getattr(instance, field)
@@ -169,7 +206,7 @@ def mutate_and_get_payload(
             )
 
         relay_id = input.get("id")
-        obj = get_obj_from_relay_id(info, relay_id)
+        obj = _get_feature(info, relay_id)
 
         target = obj.target
 

baseapp/geo/rest_framework/views.py

@@ -1,4 +1,5 @@
 import swapper
+from django.contrib.auth.models import AnonymousUser
 from django.core.exceptions import ValidationError
 from django.db.models import CharField, Expression, F, QuerySet
 from django.db.models.functions import Cast
@@ -61,6 +62,18 @@ def get_feature_properties(self) -> dict[str, Expression]:
             "feature_type": F("feature_type"),
         }
 
+    def get_cache_control(self) -> str:
+        """`public` only when anonymous users may view features.
+
+        The permission is overridable, and under a backend that hides features from
+        anonymous users a shared cache would otherwise hand one user's tile to the next
+        requester of the same z/x/y.
+        """
+        GeoJSONFeature = swapper.load_model("baseapp_geo", "GeoJSONFeature")
+        perm = f"{GeoJSONFeature._meta.app_label}.view_geojsonfeature"
+        visibility = "public" if AnonymousUser().has_perm(perm) else "private"
+        return f"{visibility}, max-age={self.cache_max_age}"
+
     def get(self, request: Request, z: int, x: int, y: int) -> HttpResponse:
         tile = Tile(z, x, y)
         if not tile.is_valid:
@@ -92,5 +105,7 @@ def get(self, request: Request, z: int, x: int, y: int) -> HttpResponse:
             content_type=MVT_CONTENT_TYPE,
             status=status.HTTP_200_OK if content else status.HTTP_204_NO_CONTENT,
         )
-        response["Cache-Control"] = f"public, max-age={self.cache_max_age}"
+        response["Cache-Control"] = self.get_cache_control()
+        # Token auth never touches the session, so nothing else would mark these apart.
+        response["Vary"] = "Authorization"
         return response

baseapp/geo/tests/test_graphql_mutations.py

@@ -1,3 +1,5 @@
+import uuid
+
 import pytest
 import swapper
 from django.contrib.auth.models import Permission
@@ -294,6 +296,79 @@ def test_perm_granted_user_deletes_feature(self, django_user_client, graphql_use
         assert GeoJSONFeature.objects.count() == 0
 
 
+class TestCrossModelRelayIds:
+    """A relay ID of another model must not be writable through these mutations.
+
+    `get_obj_from_relay_id` resolves a public ID through DocumentId and returns the object
+    of any registered model, so without a type check a user holding only the geo model
+    permissions could edit or delete rows belonging to other blocks.
+    """
+
+    def test_update_with_another_models_id_is_not_found_and_changes_nothing(
+        self, django_user_client, graphql_user_client
+    ):
+        victim = UserFactory(first_name="Untouched")
+        grant_model_perm(django_user_client.user, "change_geojsonfeature")
+
+        response = graphql_user_client(
+            UPDATE_MUTATION,
+            variables={
+                "input": {
+                    "id": victim.relay_id,
+                    "name": "Pwned",
+                    "description": "Pwned",
+                    "featureType": "pwned",
+                    "geometry": POINT_GEOJSON,
+                }
+            },
+        )
+        content = response.json()
+
+        assert content["errors"][0]["extensions"]["code"] == "not_found"
+        victim.refresh_from_db()
+        assert victim.first_name == "Untouched"
+
+    def test_delete_with_another_models_id_is_not_found_and_deletes_nothing(
+        self, django_user_client, graphql_user_client
+    ):
+        victim = UserFactory()
+        grant_model_perm(django_user_client.user, "delete_geojsonfeature")
+
+        response = graphql_user_client(
+            DELETE_MUTATION, variables={"input": {"id": victim.relay_id}}
+        )
+        content = response.json()
+
+        assert content["errors"][0]["extensions"]["code"] == "not_found"
+        assert type(victim).objects.filter(pk=victim.pk).exists()
+
+    def test_create_with_unknown_target_is_not_found_not_a_server_error(
+        self, django_user_client, graphql_user_client
+    ):
+        response = graphql_user_client(
+            CREATE_MUTATION,
+            variables={
+                "input": {
+                    "targetObjectId": str(uuid.uuid4()),
+                    "geometry": POINT_GEOJSON,
+                }
+            },
+        )
+        content = response.json()
+
+        assert content["errors"][0]["extensions"]["code"] == "not_found"
+        assert GeoJSONFeature.objects.count() == 0
+
+    def test_update_with_a_malformed_id_is_not_found(self, django_user_client, graphql_user_client):
+        grant_model_perm(django_user_client.user, "change_geojsonfeature")
+
+        response = graphql_user_client(
+            UPDATE_MUTATION, variables={"input": {"id": "not-a-relay-id", "name": "x"}}
+        )
+
+        assert response.json()["errors"][0]["extensions"]["code"] == "not_found"
+
+
 class TestWritePermissions:
     """Write-permission matrix through real mutations (AC-9.5, AC-9.6, AC-10.3)."""
 

baseapp/geo/tests/test_rest_tiles.py

@@ -202,6 +202,29 @@ def test_cache_headers(self, client) -> None:
         response = client.get(tile_url(tile_for(POINT_NYC, FEATURES_ZOOM)))
 
         assert response["Cache-Control"] == "public, max-age=300"
+        # Other Vary values are appended by DRF/middleware; only ours has to be present.
+        assert "Authorization" in response["Vary"]
+
+    def test_cache_is_private_when_features_are_not_public(
+        self, django_user_client, monkeypatch
+    ) -> None:
+        """Under a backend that hides features from anonymous users, a shared cache must
+        not be allowed to serve one user's tile to the next requester."""
+        from baseapp.geo import permissions
+
+        monkeypatch.setattr(
+            permissions.GeoPermissionsBackend,
+            "has_perm",
+            lambda self, user_obj, perm, obj=None: bool(
+                getattr(user_obj, "is_authenticated", False)
+            ),
+        )
+        GeoJSONFeatureFactory(geometry=POINT_NYC)
+
+        response = django_user_client.get(tile_url(tile_for(POINT_NYC, FEATURES_ZOOM)))
+
+        assert response.status_code == 200, response.content
+        assert response["Cache-Control"] == "private, max-age=300"
 
     def test_filters_are_the_graphql_ones(self, client) -> None:
         tree = GeoJSONFeatureFactory(geometry=POINT_NYC, feature_type="tree")