feat(master-api-key/roles): Add roles to master api key

.github/workflows/api-deploy-production-ecs.yml

@@ -39,6 +39,7 @@ jobs:
           flagsmith_saml_revision: v1.1.0
           flagsmith_workflows_revision: v1.2.5
           flagsmith_auth_controller_revision: v0.0.1
+          flagsmith_rbac_revision: v0.3.0
           flagsmith_rbac_revision: v0.3.1
 
       - name: Deploy task processor to Production

api/api_keys/authentication.py

@@ -0,0 +1,23 @@
+from contextlib import suppress
+
+from rest_framework import authentication, exceptions
+from rest_framework_api_key.permissions import KeyParser
+
+from api_keys.models import MasterAPIKey
+from api_keys.user import APIKeyUser
+
+key_parser = KeyParser()
+
+
+class MasterAPIKeyAuthentication(authentication.BaseAuthentication):
+    def authenticate(self, request):
+        key = key_parser.get(request)
+        if not key:
+            return None
+
+        with suppress(MasterAPIKey.DoesNotExist):
+            key = MasterAPIKey.objects.get_from_key(key)
+            if not key.has_expired:
+                return APIKeyUser(key), None
+
+        raise exceptions.AuthenticationFailed("Valid Master API Key not found.")

api/api_keys/migrations/0003_masterapikey_is_admin.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.20 on 2023-07-14 03:38
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api_keys', '0002_soft_delete_api_keys'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='masterapikey',
+            name='is_admin',
+            field=models.BooleanField(default=True),
+        ),
+    ]

api/api_keys/models.py

@@ -17,3 +17,4 @@ class MasterAPIKey(AbstractAPIKey, SoftDeleteObject):
     )
 
     objects = MasterAPIKeyManager()
+    is_admin = models.BooleanField(default=True)

api/api_keys/serializers.py

@@ -1,3 +1,4 @@
+from django.conf import settings
 from rest_framework import serializers
 
 from .models import MasterAPIKey
@@ -9,6 +10,7 @@ class MasterAPIKeySerializer(serializers.ModelSerializer):
         help_text="Since we don't store the api key itself(i.e: we only store the hash) this key will be none "
         "for every endpoint apart from create",
     )
+    is_admin = serializers.BooleanField(default=True)
 
     class Meta:
         model = MasterAPIKey
@@ -19,10 +21,18 @@ class Meta:
             "revoked",
             "expiry_date",
             "key",
+            "is_admin",
         )
         read_only_fields = ("prefix", "created", "key")
 
     def create(self, validated_data):
         obj, key = MasterAPIKey.objects.create_key(**validated_data)
         obj.key = key
         return obj
+
+    def validate_is_admin(self, is_admin: bool):
+        if is_admin is False and not settings.IS_RBAC_INSTALLED:
+            raise serializers.ValidationError(
+                "RBAC is not installed, cannot create non-admin key"
+            )
+        return is_admin

api/api_keys/user.py

@@ -0,0 +1,68 @@
+import typing
+
+from django.db.models import QuerySet
+
+from organisations.models import Organisation
+from permissions.permission_service import (
+    get_permitted_environments_for_master_api_key,
+    get_permitted_projects_for_master_api_key,
+    is_master_api_key_environment_admin,
+    is_master_api_key_project_admin,
+    master_api_key_has_organisation_permission,
+)
+from users.abc import UserABC
+
+from .models import MasterAPIKey
+
+if typing.TYPE_CHECKING:
+    from environments.models import Environment
+    from projects.models import Project
+
+
+class APIKeyUser(UserABC):
+    def __init__(self, key: MasterAPIKey):
+        self.key = key
+
+    @property
+    def is_authenticated(self) -> bool:
+        return True
+
+    @property
+    def is_master_api_key_user(self) -> bool:
+        return True
+
+    def belongs_to(self, organisation_id: int) -> bool:
+        return self.key.organisation_id == organisation_id
+
+    def is_project_admin(self, project: "Project") -> bool:
+        return is_master_api_key_project_admin(self.key, project)
+
+    def is_environment_admin(self, environment: "Environment") -> bool:
+        return is_master_api_key_environment_admin(self.key, environment)
+
+    def has_project_permission(self, permission: str, project: "Project") -> bool:
+        return project in self.get_permitted_projects(permission)
+
+    def has_environment_permission(
+        self, permission: str, environment: "Environment"
+    ) -> bool:
+        return environment in self.get_permitted_environments(
+            permission, environment.project
+        )
+
+    def has_organisation_permission(
+        self, organisation: Organisation, permission_key: str
+    ) -> bool:
+        return master_api_key_has_organisation_permission(
+            self.key, organisation, permission_key
+        )
+
+    def get_permitted_projects(self, permission_key: str) -> QuerySet["Project"]:
+        return get_permitted_projects_for_master_api_key(self.key, permission_key)
+
+    def get_permitted_environments(
+        self, permission_key: str, project: "Project"
+    ) -> QuerySet["Environment"]:
+        return get_permitted_environments_for_master_api_key(
+            self.key, project, permission_key
+        )

api/app/settings/common.py

@@ -221,6 +221,7 @@
     "DEFAULT_PERMISSION_CLASSES": ["rest_framework.permissions.IsAuthenticated"],
     "DEFAULT_AUTHENTICATION_CLASSES": (
         "rest_framework.authentication.TokenAuthentication",
+        "api_keys.authentication.MasterAPIKeyAuthentication",
     ),
     "PAGE_SIZE": 10,
     "UNICODE_JSON": False,
@@ -248,8 +249,6 @@
     "django.contrib.messages.middleware.MessageMiddleware",
     "django.middleware.clickjacking.XFrameOptionsMiddleware",
     "simple_history.middleware.HistoryRequestMiddleware",
-    # Add master api key object to request
-    "api_keys.middleware.MasterAPIKeyMiddleware",
 ]
 
 ADD_NEVER_CACHE_HEADERS = env.bool("ADD_NEVER_CACHE_HEADERS", True)

api/app/settings/test.py

@@ -1,21 +1,12 @@
 from app.settings.common import *  # noqa
+from app.settings.common import REST_FRAMEWORK
 
 # We dont want to track tests
 ENABLE_TELEMETRY = False
 
-REST_FRAMEWORK = {
-    "DEFAULT_PERMISSION_CLASSES": ["rest_framework.permissions.IsAuthenticated"],
-    "DEFAULT_AUTHENTICATION_CLASSES": (
-        "rest_framework.authentication.TokenAuthentication",
-    ),
-    "PAGE_SIZE": 10,
-    "UNICODE_JSON": False,
-    "DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
-    "DEFAULT_THROTTLE_RATES": {
-        "login": "100/min",
-        "mfa_code": "5/min",
-        "invite": "10/min",
-        "signup": "100/min",
-    },
-    "DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"],
+REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"] = {
+    "login": "100/min",
+    "mfa_code": "5/min",
+    "invite": "10/min",
+    "signup": "100/min",
 }

api/conftest.py

@@ -292,19 +292,36 @@ def environment_api_key(environment):
 
 
 @pytest.fixture()
-def master_api_key(organisation) -> Tuple[MasterAPIKey, str]:
-    master_api_key, key = MasterAPIKey.objects.create_key(
+def admin_master_api_key(organisation) -> typing.Tuple[MasterAPIKey, str]:
+    master_api_key, _ = MasterAPIKey.objects.create_key(
+        name="test_key", organisation=organisation, is_admin=True
+    )
+    return master_api_key
+
+
+@pytest.fixture()
+def master_api_key(organisation) -> typing.Tuple[MasterAPIKey, str]:
+    master_api_key, _ = MasterAPIKey.objects.create_key(
+        name="test_key", organisation=organisation, is_admin=False
+    )
+    return master_api_key
+
+
+@pytest.fixture()
+def master_api_key_and_obj(organisation) -> typing.Tuple[MasterAPIKey, str]:
+    master_api_key_obj, key = MasterAPIKey.objects.create_key(
         name="test_key", organisation=organisation
     )
-    return master_api_key, key
+    return key, master_api_key_obj
 
 
 @pytest.fixture()
-def master_api_key_client(master_api_key):
+def master_api_key_client(master_api_key_and_obj):
+    key = master_api_key_and_obj[0]
     # Can not use `api_client` fixture here because:
     # https://docs.pytest.org/en/6.2.x/fixture.html#fixtures-can-be-requested-more-than-once-per-test-return-values-are-cached
     api_client = APIClient()
-    api_client.credentials(HTTP_AUTHORIZATION="Api-Key " + master_api_key[1])
+    api_client.credentials(HTTP_AUTHORIZATION="Api-Key " + key)
     return api_client
 
 

api/core/models.py

@@ -4,6 +4,7 @@
 
 from django.db import models
 from django.db.models import Manager
+from django.http import HttpRequest
 from simple_history.models import HistoricalRecords
 from softdelete.models import SoftDeleteManager, SoftDeleteObject
 
@@ -132,13 +133,21 @@ def _get_project(self) -> typing.Optional["Project"]:
         return None
 
 
+def get_history_user(
+    instance: typing.Any, request: HttpRequest
+) -> typing.Optional["FFAdminUser"]:
+    user = getattr(request, "user", None)
+    return None if getattr(user, "is_master_api_key_user", False) else user
+
+
 def abstract_base_auditable_model_factory(
     historical_records_excluded_fields: typing.List[str] = None,
 ) -> typing.Type[_AbstractBaseAuditableModel]:
     class Base(_AbstractBaseAuditableModel):
         history = HistoricalRecords(
             bases=[BaseHistoricalModel],
             excluded_fields=historical_records_excluded_fields or [],
+            get_user=get_history_user,
             inherit=True,
         )
 

api/core/signals.py

@@ -43,8 +43,7 @@ def create_audit_log_from_historical_record(
 def add_master_api_key(sender, **kwargs):
     try:
         history_instance = kwargs["history_instance"]
-        history_instance.master_api_key = (
-            HistoricalRecords.thread.request.master_api_key
-        )
+        master_api_key = HistoricalRecords.thread.request.user.key
+        history_instance.master_api_key = master_api_key
     except (KeyError, AttributeError):
         pass

api/environments/permissions/permissions.py

@@ -1,7 +1,6 @@
 import typing
 
 from django.db.models import Model, Q
-from django.http import HttpRequest
 from rest_framework import exceptions
 from rest_framework.permissions import BasePermission, IsAuthenticated
 
@@ -44,9 +43,6 @@ def has_permission(self, request, view):
         return True
 
     def has_object_permission(self, request, view, obj):
-        if request.user.is_anonymous:
-            return False
-
         if view.action == "clone":
             return request.user.is_project_admin(obj.project)
 
@@ -55,36 +51,6 @@ def has_object_permission(self, request, view, obj):
         ]
 
 
-class MasterAPIKeyEnvironmentPermissions(BasePermission):
-    def has_permission(self, request: HttpRequest, view: str) -> bool:
-        master_api_key = getattr(request, "master_api_key", None)
-
-        if not master_api_key:
-            return False
-
-        if view.action == "create":
-            try:
-                project_id = request.data.get("project")
-                project = Project.objects.get(id=project_id)
-                return master_api_key.organisation_id == project.organisation.id
-
-            except Project.DoesNotExist:
-                return False
-
-        # return true as list will be handled by view and obj permissions will be handled later
-        return True
-
-    def has_object_permission(
-        self, request: HttpRequest, view: str, obj: Model
-    ) -> bool:
-        master_api_key = getattr(request, "master_api_key", None)
-
-        if not master_api_key:
-            return False
-
-        return master_api_key.organisation_id == obj.project.organisation_id
-
-
 class IdentityPermissions(BasePermission):
     def has_permission(self, request, view):
         try: