feat: Exclude verified Edge Proxy requests from API usage tracking

api/app_analytics/middleware.py

@@ -1,5 +1,8 @@
+import importlib
 from typing import Callable
 
+from common.core.utils import is_saas
+from django.conf import settings
 from django.http import HttpRequest, HttpResponse
 
 from app_analytics.mappers import map_request_to_labels
@@ -30,9 +33,27 @@ def __init__(
         get_response: Callable[[HttpRequest], HttpResponse],
     ) -> None:
         self.get_response = get_response
+        # An Edge Proxy reports the requests it serves itself, so its own
+        # requests to core must not also be counted here. The private
+        # edge_proxy app decides what counts as the proxy's own: a verified
+        # X-Proxy-Key whose grants cover the presented environment — bare
+        # header presence is never trusted.
+        self.is_edge_proxy_request: Callable[[HttpRequest], bool] | None = None
+        if settings.EDGE_PROXY_INSTALLED and not is_saas():
+            # getattr, not a hard import: the installed edge_proxy app may
+            # predate the helper — the proxy's fetches are then counted as
+            # before.
+            self.is_edge_proxy_request = getattr(
+                importlib.import_module("edge_proxy.authentication"),
+                "is_edge_proxy_request",
+                None,
+            )
 
     def __call__(self, request: HttpRequest) -> HttpResponse:
-        if environment_key := request.headers.get("X-Environment-Key"):
+        if (environment_key := request.headers.get("X-Environment-Key")) and not (
+            self.is_edge_proxy_request is not None
+            and self.is_edge_proxy_request(request)
+        ):
             track_usage_by_resource_host_and_environment(
                 resource=get_resource_from_uri(request.path),
                 host=request.get_host(),

api/tests/unit/app_analytics/test_middleware.py

@@ -1,3 +1,5 @@
+import sys
+
 import pytest
 from django.test import RequestFactory
 from pytest_django.fixtures import SettingsWrapper
@@ -8,6 +10,13 @@
 from tests.types import EnableFeaturesFixture
 
 
+@pytest.fixture(autouse=True)
+def edge_proxy_not_installed(settings: SettingsWrapper) -> None:
+    # Keep these tests hermetic: whether the private edge_proxy wheel is
+    # installed in the test environment must not change middleware wiring.
+    settings.EDGE_PROXY_INSTALLED = False
+
+
 @pytest.mark.parametrize(
     "path, resource_name",
     [
@@ -131,3 +140,113 @@ def test_api_usage_middleware__request_not_tracked__not_calls_expected(
 
     # Then
     mocked_track_request.delay.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "edge_proxy_installed, saas, expect_wired",
+    [
+        (True, False, True),
+        (True, True, False),
+        (False, False, False),
+    ],
+)
+def test_api_usage_middleware__edge_proxy_check__wired_only_where_expected(
+    mocker: MockerFixture,
+    settings: SettingsWrapper,
+    edge_proxy_installed: bool,
+    saas: bool,
+    expect_wired: bool,
+) -> None:
+    # Given a deployment with/without the private edge_proxy app
+    settings.EDGE_PROXY_INSTALLED = edge_proxy_installed
+    mocker.patch("app_analytics.middleware.is_saas", return_value=saas)
+    is_edge_proxy_request = mocker.MagicMock()
+    mocker.patch.dict(
+        sys.modules,
+        {
+            "edge_proxy": mocker.MagicMock(),
+            "edge_proxy.authentication": mocker.MagicMock(
+                is_edge_proxy_request=is_edge_proxy_request
+            ),
+        },
+    )
+
+    # When
+    middleware = APIUsageMiddleware(mocker.MagicMock())
+
+    # Then the verifier is wired only where the proxy reports usage
+    # itself: a non-SaaS deployment with the edge_proxy app installed
+    assert middleware.is_edge_proxy_request is (
+        is_edge_proxy_request if expect_wired else None
+    )
+
+
+def test_api_usage_middleware__edge_proxy_app_predates_helper__not_wired(
+    mocker: MockerFixture,
+    settings: SettingsWrapper,
+) -> None:
+    # Given an installed edge_proxy app without is_edge_proxy_request
+    settings.EDGE_PROXY_INSTALLED = True
+    mocker.patch("app_analytics.middleware.is_saas", return_value=False)
+    mocker.patch.dict(
+        sys.modules,
+        {
+            "edge_proxy": mocker.MagicMock(),
+            "edge_proxy.authentication": mocker.MagicMock(spec=[]),
+        },
+    )
+
+    # When
+    middleware = APIUsageMiddleware(mocker.MagicMock())
+
+    # Then requests are counted as before
+    assert middleware.is_edge_proxy_request is None
+
+
+@pytest.mark.parametrize("is_verified_proxy_request", [True, False])
+def test_api_usage_middleware__edge_proxy_check_wired__tracks_unverified_only(
+    rf: RequestFactory,
+    mocker: MockerFixture,
+    settings: SettingsWrapper,
+    is_verified_proxy_request: bool,
+) -> None:
+    # Given a request bearing proxy headers the edge_proxy app does or
+    # does not verify
+    headers = {"HTTP_X-Environment-Key": "test", "HTTP_X-Proxy-Key": "pk.key"}
+    request = rf.get("/api/v1/environment-document", **headers)  # type: ignore[arg-type]
+    settings.EDGE_PROXY_INSTALLED = False
+    mocked_track_usage = mocker.patch(
+        "app_analytics.middleware.track_usage_by_resource_host_and_environment"
+    )
+    is_edge_proxy_request = mocker.MagicMock(return_value=is_verified_proxy_request)
+    middleware = APIUsageMiddleware(mocker.MagicMock())
+    middleware.is_edge_proxy_request = is_edge_proxy_request
+
+    # When
+    middleware(request)
+
+    # Then only a verified proxy request is exempt — a spoofed header is not
+    is_edge_proxy_request.assert_called_once_with(request)
+    assert mocked_track_usage.called is not is_verified_proxy_request
+
+
+def test_api_usage_middleware__edge_proxy_check_not_wired__proxy_header_still_tracked(
+    rf: RequestFactory,
+    mocker: MockerFixture,
+    settings: SettingsWrapper,
+) -> None:
+    # Given a request bearing an X-Proxy-Key on a deployment with no
+    # edge_proxy app to verify it
+    headers = {"HTTP_X-Environment-Key": "test", "HTTP_X-Proxy-Key": "pk.key"}
+    request = rf.get("/api/v1/environment-document", **headers)  # type: ignore[arg-type]
+    settings.EDGE_PROXY_INSTALLED = False
+    mocked_track_usage = mocker.patch(
+        "app_analytics.middleware.track_usage_by_resource_host_and_environment"
+    )
+
+    # When
+    middleware = APIUsageMiddleware(mocker.MagicMock())
+    middleware(request)
+
+    # Then
+    mocked_track_usage.assert_called_once()