feat(cdp): disallow hog code modification

posthog/api/hog_function.py

@@ -37,7 +37,7 @@
 )
 from posthog.models.plugin import TranspilerError
 from posthog.plugins.plugin_server_api import create_hog_invocation_test
-
+from django.conf import settings
 
 logger = structlog.get_logger(__name__)
 
@@ -192,6 +192,16 @@ def validate(self, attrs):
                 attrs["inputs_schema"] = attrs.get("inputs_schema") or template.inputs_schema
                 attrs["inputs"] = attrs.get("inputs") or {}
 
+        if hog_type == "transformation":
+            if not settings.HOG_TRANSFORMATIONS_CUSTOM_HOG_ENABLED:
+                if not template:
+                    raise serializers.ValidationError(
+                        {"template_id": "Transformation functions must be created from a template."}
+                    )
+                # Currently we do not allow modifying the core transformation templates when transformations are disabled
+                attrs["hog"] = template.hog
+                attrs["inputs_schema"] = template.inputs_schema
+
         # Used for both top level input validation, and mappings input validation
         def validate_input_and_filters(attrs: dict):
             if "inputs_schema" in attrs:

posthog/api/test/test_hog_function.py

@@ -6,6 +6,7 @@
 from freezegun import freeze_time
 from inline_snapshot import snapshot
 from rest_framework import status
+from django.test.utils import override_settings
 
 from common.hogvm.python.operation import HOGQL_BYTECODE_VERSION
 from posthog.api.test.test_hog_function_templates import MOCK_NODE_TEMPLATES
@@ -1264,6 +1265,127 @@ def create(payload):
             }
         )
 
+    @override_settings(HOG_TRANSFORMATIONS_CUSTOM_HOG_ENABLED=False)
+    def test_transformation_functions_require_template_when_disabled(self):
+        response = self.client.post(
+            f"/api/projects/{self.team.id}/hog_functions/",
+            data={
+                "name": "Custom Transform",
+                "type": "transformation",
+                "hog": "return event",
+            },
+        )
+
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.json() == {
+            "type": "validation_error",
+            "code": "invalid_input",
+            "detail": "Transformation functions must be created from a template.",
+            "attr": "template_id",
+        }
+
+    @override_settings(HOG_TRANSFORMATIONS_CUSTOM_HOG_ENABLED=False)
+    def test_transformation_functions_preserve_template_code_when_disabled(self):
+        with patch("posthog.api.hog_function_template.HogFunctionTemplates.template") as mock_template:
+            mock_template.return_value = template_slack  # Use existing template instead of creating mock
+
+            # First create with transformations enabled
+            with override_settings(HOG_TRANSFORMATIONS_CUSTOM_HOG_ENABLED=True):
+                response = self.client.post(
+                    f"/api/projects/{self.team.id}/hog_functions/",
+                    data={
+                        "name": "Template Transform",
+                        "type": "transformation",
+                        "template_id": template_slack.id,
+                        "hog": "return modified_event",
+                        "inputs": {
+                            "slack_workspace": {"value": 1},
+                            "channel": {"value": "#general"},
+                        },
+                    },
+                )
+                assert response.status_code == status.HTTP_201_CREATED, response.json()
+                function_id = response.json()["id"]
+
+            # Try to update with transformations disabled
+            response = self.client.patch(
+                f"/api/projects/{self.team.id}/hog_functions/{function_id}/",
+                data={
+                    "hog": "return another_event",
+                },
+            )
+
+            assert response.status_code == status.HTTP_200_OK
+            assert response.json()["hog"] == template_slack.hog  # Original template code preserved
+
+    @override_settings(HOG_TRANSFORMATIONS_ENABLED=True)
+    def test_transformation_uses_template_code_even_when_enabled(self):
+        # Even with transformations enabled, we should still use template code
+        response = self.client.post(
+            f"/api/projects/{self.team.id}/hog_functions/",
+            data={
+                "type": "transformation",
+                "name": "Test Function",
+                "description": "Test description",
+                "template_id": template_slack.id,
+                "hog": "return custom_event",  # This should be ignored
+                "inputs": {
+                    "slack_workspace": {"value": 1},
+                    "channel": {"value": "#general"},
+                },
+            },
+        )
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["hog"] == template_slack.hog  # Should always use template code
+
+    def test_transformation_type_gets_execution_order_automatically(self):
+        with patch("posthog.api.hog_function_template.HogFunctionTemplates.template") as mock_template:
+            mock_template.return_value = template_slack
+
+            # Create first transformation function
+            response1 = self.client.post(
+                f"/api/projects/{self.team.id}/hog_functions/",
+                data={
+                    "type": "transformation",
+                    "name": "First Transformation",
+                    "template_id": template_slack.id,
+                    "inputs": {
+                        "slack_workspace": {"value": 1},
+                        "channel": {"value": "#general"},
+                    },
+                },
+            )
+            assert response1.status_code == status.HTTP_201_CREATED
+            assert response1.json()["execution_order"] == 1
+
+            # Create second transformation function
+            response2 = self.client.post(
+                f"/api/projects/{self.team.id}/hog_functions/",
+                data={
+                    "type": "transformation",
+                    "name": "Second Transformation",
+                    "template_id": template_slack.id,
+                    "inputs": {
+                        "slack_workspace": {"value": 1},
+                        "channel": {"value": "#general"},
+                    },
+                },
+            )
+            assert response2.status_code == status.HTTP_201_CREATED
+            assert response2.json()["execution_order"] == 2
+
+            # Create a non-transformation function - should not get execution_order
+            response3 = self.client.post(
+                f"/api/projects/{self.team.id}/hog_functions/",
+                data={
+                    **EXAMPLE_FULL,  # This is fine for destination type
+                    "type": "destination",
+                    "name": "Destination Function",
+                },
+            )
+            assert response3.status_code == status.HTTP_201_CREATED
+            assert response3.json()["execution_order"] is None
+
     def test_list_hog_functions_ordered_by_execution_order_and_created_at(self):
         # Create functions with different execution orders and creation times
         with freeze_time("2024-01-01T00:00:00Z"):
@@ -1318,40 +1440,3 @@ def test_list_hog_functions_ordered_by_execution_order_and_created_at(self):
             "Function 3",  # execution_order=2
             "Function 4",  # execution_order=null
         ]
-
-    def test_transformation_type_gets_execution_order_automatically(self):
-        # Create first transformation function
-        response1 = self.client.post(
-            f"/api/projects/{self.team.id}/hog_functions/",
-            data={
-                **EXAMPLE_FULL,
-                "type": "transformation",
-                "name": "First Transformation",
-            },
-        )
-        assert response1.status_code == status.HTTP_201_CREATED
-        assert response1.json()["execution_order"] == 1
-
-        # Create second transformation function
-        response2 = self.client.post(
-            f"/api/projects/{self.team.id}/hog_functions/",
-            data={
-                **EXAMPLE_FULL,
-                "type": "transformation",
-                "name": "Second Transformation",
-            },
-        )
-        assert response2.status_code == status.HTTP_201_CREATED
-        assert response2.json()["execution_order"] == 2
-
-        # Create a non-transformation function - should not get execution_order
-        response3 = self.client.post(
-            f"/api/projects/{self.team.id}/hog_functions/",
-            data={
-                **EXAMPLE_FULL,
-                "type": "destination",
-                "name": "Destination Function",
-            },
-        )
-        assert response3.status_code == status.HTTP_201_CREATED
-        assert response3.json()["execution_order"] is None

posthog/settings/web.py

@@ -416,3 +416,8 @@ def static_varies_origin(headers, path, url):
 REMOTE_CONFIG_CDN_PURGE_ENDPOINT = get_from_env("REMOTE_CONFIG_CDN_PURGE_ENDPOINT", "")
 REMOTE_CONFIG_CDN_PURGE_TOKEN = get_from_env("REMOTE_CONFIG_CDN_PURGE_TOKEN", "")
 REMOTE_CONFIG_CDN_PURGE_DOMAINS = get_list(os.getenv("REMOTE_CONFIG_CDN_PURGE_DOMAINS", ""))
+
+# Whether to allow modification of transformation code
+HOG_TRANSFORMATIONS_CUSTOM_HOG_ENABLED = get_from_env(
+    "HOG_TRANSFORMATIONS_CUSTOM_HOG_ENABLED", False, type_cast=str_to_bool
+)