fix(Audit): Retrieving records via /api/v1/audit/ with a master API key results in error

api/api_keys/user.py

@@ -42,6 +42,11 @@ def is_master_api_key_user(self) -> bool:
     def organisations(self) -> QuerySet[Organisation]:
         return Organisation.objects.filter(id=self.key.organisation_id)  # type: ignore[no-any-return]
 
+    def get_admin_organisations(self) -> QuerySet[Organisation]:
+        if not self.key.is_admin:
+            return Organisation.objects.none()  # type: ignore[no-any-return]
+        return Organisation.objects.filter(id=self.key.organisation_id)  # type: ignore[no-any-return]
+
     def belongs_to(self, organisation_id: int) -> bool:
         return self.key.organisation_id == organisation_id
 

api/audit/views.py

@@ -19,7 +19,7 @@
     AuditLogRetrieveSerializer,
     AuditLogsQueryParamSerializer,
 )
-from organisations.models import Organisation, OrganisationRole
+from organisations.models import Organisation
 
 
 @method_decorator(
@@ -93,8 +93,7 @@ def _get_organisation(self) -> Organisation | None:
 class AllAuditLogViewSet(_BaseAuditLogViewSet):
     def _get_base_filters(self) -> Q:
         return Q(
-            project__organisation__userorganisation__user=self.request.user,
-            project__organisation__userorganisation__role=OrganisationRole.ADMIN,
+            project__organisation__in=self.request.user.get_admin_organisations()  # type: ignore[union-attr]
         )
 
     def _get_organisation(self) -> Organisation | None:
@@ -109,9 +108,7 @@ def _get_organisation(self) -> Organisation | None:
         Since we're applying the base filters to the query set
         """
         return (  # type: ignore[no-any-return]
-            self.request.user.organisations.filter(  # type: ignore[union-attr]
-                userorganisation__role=OrganisationRole.ADMIN
-            )
+            self.request.user.get_admin_organisations()  # type: ignore[union-attr]
             .select_related("subscription", "subscription_information_cache")
             .first()
         )

api/tests/unit/audit/test_unit_audit_views.py

@@ -9,6 +9,7 @@
 from rest_framework import status
 from rest_framework.test import APIClient
 
+from api_keys.models import MasterAPIKey
 from audit.constants import ENVIRONMENT_FEATURE_VERSION_PUBLISHED_MESSAGE
 from audit.models import AuditLog
 from audit.related_object_type import RelatedObjectType
@@ -171,6 +172,69 @@ def test_list_audit_log__admin_of_another_organisation__returns_empty(
     assert response.json()["count"] == 0
 
 
+def test_list_audit_log__admin_master_api_key__returns_organisation_logs(
+    admin_master_api_key_client: APIClient,
+    organisation: Organisation,
+    project: Project,
+) -> None:
+    # Given
+    audit_log = AuditLog.objects.create(project=project)
+    url = reverse("api-v1:audit-list")
+
+    # When
+    response = admin_master_api_key_client.get(url)
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    response_json = response.json()
+    assert response_json["count"] == 1
+    assert response_json["results"][0]["id"] == audit_log.id
+
+
+def test_list_audit_log__non_admin_master_api_key__returns_empty(
+    master_api_key: typing.Tuple[MasterAPIKey, str],
+    api_client: APIClient,
+    organisation: Organisation,
+    project: Project,
+) -> None:
+    # Given
+    AuditLog.objects.create(project=project)
+    url = reverse("api-v1:audit-list")
+
+    api_client.credentials(HTTP_AUTHORIZATION="Api-Key " + master_api_key[1])
+
+    # When
+    response = api_client.get(url)
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["count"] == 0
+
+
+def test_list_audit_log__master_api_key_of_another_organisation__returns_empty(
+    api_client: APIClient,
+    organisation: Organisation,
+    project: Project,
+) -> None:
+    # Given
+    another_organisation = Organisation.objects.create(name="another organisation")
+    _, key = MasterAPIKey.objects.create_key(
+        name="another_key", organisation=another_organisation, is_admin=True
+    )
+
+    AuditLog.objects.create(project=project)
+    url = reverse("api-v1:audit-list")
+
+    api_client.credentials(HTTP_AUTHORIZATION="Api-Key " + key)
+
+    # When
+    response = api_client.get(url)
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json()["count"] == 0
+
+
 def test_retrieve_audit_log__environment_feature_version_published__includes_required_fields(
     admin_client: APIClient,
     admin_user: FFAdminUser,

api/tests/unit/platform_hub/test_views.py

@@ -3,6 +3,7 @@
 from rest_framework import status
 from rest_framework.test import APIClient
 
+from api_keys.models import MasterAPIKey
 from environments.models import Environment
 from features.models import Feature
 from organisations.models import Organisation, OrganisationRole
@@ -146,6 +147,39 @@ def test_organisations_view__user_admin_of_one_org__returns_only_that_org(
     assert data[0]["name"] == "Platform Hub Org"
 
 
+def test_organisations_view__admin_master_api_key__returns_only_that_org(
+    platform_hub_organisation: Organisation,
+    platform_hub_project: Project,
+    platform_hub_environment: Environment,
+    platform_hub_feature: Feature,
+    other_organisation: Organisation,
+    other_org_project: Project,
+    other_org_environment: Environment,
+    other_org_feature: Feature,
+    settings: pytest.FixtureRequest,
+) -> None:
+    # Given
+    settings.USE_POSTGRES_FOR_ANALYTICS = False  # type: ignore[attr-defined]
+    settings.INFLUXDB_TOKEN = ""  # type: ignore[attr-defined]
+
+    _, key = MasterAPIKey.objects.create_key(
+        name="platform_hub_key", organisation=platform_hub_organisation, is_admin=True
+    )
+    client = APIClient()
+    client.credentials(HTTP_AUTHORIZATION="Api-Key " + key)
+
+    url = reverse("api-v1:platform-hub:organisations")
+
+    # When
+    response = client.get(url)
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()
+    assert len(data) == 1
+    assert data[0]["id"] == platform_hub_organisation.id
+
+
 def test_organisations_view__non_admin_org_exists__excludes_non_admin_orgs(
     platform_hub_admin_client: APIClient,
     platform_hub_organisation: Organisation,

api/tests/unit/users/test_unit_users_models.py

@@ -73,7 +73,7 @@ def test_get_admin_organisations__user_with_mixed_roles__returns_only_admin_orgs
     admin_user.add_organisation(non_admin_organisation, OrganisationRole.USER)
 
     # When
-    admin_orgs = admin_user.get_admin_organisations()  # type: ignore[no-untyped-call]
+    admin_orgs = admin_user.get_admin_organisations()
 
     # Then
     assert organisation in admin_orgs

api/users/abc.py

@@ -17,6 +17,10 @@ def is_authenticated(self) -> bool:
     def belongs_to(self, organisation_id: int) -> bool:
         raise NotImplementedError()
 
+    @abstractmethod
+    def get_admin_organisations(self) -> QuerySet[Organisation]:
+        raise NotImplementedError()
+
     @abstractmethod
     def is_project_admin(self, project: "Project") -> bool:
         raise NotImplementedError()

api/users/models.py

@@ -254,8 +254,8 @@ def join_organisation_from_invite(self, invite: "AbstractBaseInviteModel"):  # t
     def is_organisation_admin(self, organisation: typing.Union["Organisation", int]):  # type: ignore[no-untyped-def]
         return is_user_organisation_admin(self, organisation)
 
-    def get_admin_organisations(self):  # type: ignore[no-untyped-def]
-        return Organisation.objects.filter(
+    def get_admin_organisations(self) -> QuerySet[Organisation]:
+        return Organisation.objects.filter(  # type: ignore[no-any-return]
             userorganisation__user=self,
             userorganisation__role=OrganisationRole.ADMIN.name,
         )