♻️(backend) replace Django permissions with feature flag decorator

Refactor feature flag mechanism from Django permission classes to custom
decorator that returns 404 Not Found when features are disabled instead
of exposing API structure through permission errors.

Improves security by preventing information disclosure about disabled
features and provides more appropriate response semantics. Custom
decorator approach is better suited for feature toggling than Django's
permission system which is designed for authorization.
This commit is contained in:
lebaudantoine
2025-09-08 13:17:03 +02:00
committed by aleb_the_flash
parent 58722cab00
commit 8044e3d6d8
7 changed files with 62 additions and 45 deletions

View File

@@ -0,0 +1,45 @@
"""Feature flag handler for the Meet core app."""
from functools import wraps
from django.conf import settings
from django.http import Http404
class FeatureFlag:
"""Check if features are enabled and return error responses."""
FLAGS = {
"recording": "RECORDING_ENABLE",
"storage_event": "RECORDING_STORAGE_EVENT_ENABLE",
"subtitle": "ROOM_SUBTITLE_ENABLED",
}
@classmethod
def flag_is_active(cls, flag_name):
"""Check if a feature flag is active."""
setting_name = cls.FLAGS.get(flag_name)
if setting_name is None:
return False
return getattr(settings, setting_name, False)
@classmethod
def require(cls, flag_name):
"""Decorator to check feature at the beginning of endpoint methods."""
if flag_name not in cls.FLAGS:
raise ValueError(f"Unknown feature flag: {flag_name}")
def decorator(view_func):
@wraps(view_func)
def wrapper(self, request, *args, **kwargs):
if not cls.flag_is_active(flag_name):
raise Http404
return view_func(self, request, *args, **kwargs)
return wrapper
return decorator

View File

@@ -1,7 +1,5 @@
"""Permission handlers for the Meet core app.""" """Permission handlers for the Meet core app."""
from django.conf import settings
from rest_framework import permissions from rest_framework import permissions
from ..models import RoleChoices from ..models import RoleChoices
@@ -101,36 +99,6 @@ class HasPrivilegesOnRoom(IsAuthenticated):
return obj.is_administrator_or_owner(request.user) return obj.is_administrator_or_owner(request.user)
class IsRecordingEnabled(permissions.BasePermission):
"""Check if the recording feature is enabled."""
message = "Access denied, recording is disabled."
def has_permission(self, request, view):
"""Determine if access is allowed based on settings."""
return settings.RECORDING_ENABLE
class IsStorageEventEnabled(permissions.BasePermission):
"""Check if the storage event feature is enabled."""
message = "Access denied, storage event is disabled."
def has_permission(self, request, view):
"""Determine if access is allowed based on settings."""
return settings.RECORDING_STORAGE_EVENT_ENABLE
class IsSubtitleEnabled(permissions.BasePermission):
"""Check if the subtitle feature is enabled."""
message = "Access denied, subtitles are disabled."
def has_permission(self, request, view):
"""Determine if access is allowed based on settings."""
return settings.ROOM_SUBTITLE_ENABLED
class HasLiveKitRoomAccess(permissions.BasePermission): class HasLiveKitRoomAccess(permissions.BasePermission):
"""Check if authenticated user's LiveKit token is for the specific room.""" """Check if authenticated user's LiveKit token is for the specific room."""

View File

@@ -59,6 +59,7 @@ from core.services.subtitle import SubtitleException, SubtitleService
from ..authentication.livekit import LiveKitTokenAuthentication from ..authentication.livekit import LiveKitTokenAuthentication
from . import permissions, serializers from . import permissions, serializers
from .feature_flag import FeatureFlag
# pylint: disable=too-many-ancestors # pylint: disable=too-many-ancestors
@@ -293,9 +294,9 @@ class RoomViewSet(
url_path="start-recording", url_path="start-recording",
permission_classes=[ permission_classes=[
permissions.HasPrivilegesOnRoom, permissions.HasPrivilegesOnRoom,
permissions.IsRecordingEnabled,
], ],
) )
@FeatureFlag.require("recording")
def start_room_recording(self, request, pk=None): # pylint: disable=unused-argument def start_room_recording(self, request, pk=None): # pylint: disable=unused-argument
"""Start recording a room.""" """Start recording a room."""
@@ -338,9 +339,9 @@ class RoomViewSet(
url_path="stop-recording", url_path="stop-recording",
permission_classes=[ permission_classes=[
permissions.HasPrivilegesOnRoom, permissions.HasPrivilegesOnRoom,
permissions.IsRecordingEnabled,
], ],
) )
@FeatureFlag.require("recording")
def stop_room_recording(self, request, pk=None): # pylint: disable=unused-argument def stop_room_recording(self, request, pk=None): # pylint: disable=unused-argument
"""Stop room recording.""" """Stop room recording."""
@@ -542,11 +543,11 @@ class RoomViewSet(
methods=["post"], methods=["post"],
url_path="start-subtitle", url_path="start-subtitle",
permission_classes=[ permission_classes=[
permissions.IsSubtitleEnabled,
permissions.HasLiveKitRoomAccess, permissions.HasLiveKitRoomAccess,
], ],
authentication_classes=[LiveKitTokenAuthentication], authentication_classes=[LiveKitTokenAuthentication],
) )
@FeatureFlag.require("subtitle")
def start_subtitle(self, request, pk=None): # pylint: disable=unused-argument def start_subtitle(self, request, pk=None): # pylint: disable=unused-argument
"""Start realtime transcription for the room. """Start realtime transcription for the room.
@@ -732,8 +733,8 @@ class RecordingViewSet(
methods=["post"], methods=["post"],
url_path="storage-hook", url_path="storage-hook",
authentication_classes=[StorageEventAuthentication], authentication_classes=[StorageEventAuthentication],
permission_classes=[permissions.IsStorageEventEnabled],
) )
@FeatureFlag.require("storage_event")
def on_storage_event_received(self, request, pk=None): # pylint: disable=unused-argument def on_storage_event_received(self, request, pk=None): # pylint: disable=unused-argument
"""Handle incoming storage hook events for recordings.""" """Handle incoming storage hook events for recordings."""

View File

@@ -77,7 +77,8 @@ def test_save_recording_permission_needed(settings, client):
HTTP_AUTHORIZATION="Bearer testAuthToken", HTTP_AUTHORIZATION="Bearer testAuthToken",
) )
assert response.status_code == 403 assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
def test_save_recording_parsing_error(recording_settings, mock_get_parser, client): def test_save_recording_parsing_error(recording_settings, mock_get_parser, client):

View File

@@ -55,8 +55,9 @@ def test_start_recording_anonymous():
assert Recording.objects.count() == 0 assert Recording.objects.count() == 0
def test_start_recording_non_owner_and_non_administrator(): def test_start_recording_non_owner_and_non_administrator(settings):
"""Non-owner and Non-Administrator users should not be allowed to start room recordings.""" """Non-owner and Non-Administrator users should not be allowed to start room recordings."""
settings.RECORDING_ENABLE = True
room = RoomFactory() room = RoomFactory()
user = UserFactory() user = UserFactory()
client = APIClient() client = APIClient()
@@ -88,8 +89,8 @@ def test_start_recording_recording_disabled(settings):
{"mode": "screen_recording"}, {"mode": "screen_recording"},
) )
assert response.status_code == 403 assert response.status_code == 404
assert response.json() == {"detail": "Access denied, recording is disabled."} assert response.json() == {"detail": "Not found."}
assert Recording.objects.count() == 0 assert Recording.objects.count() == 0

View File

@@ -54,8 +54,9 @@ def test_stop_recording_anonymous():
assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1 assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
def test_stop_recording_non_owner_and_non_administrator(): def test_stop_recording_non_owner_and_non_administrator(settings):
"""Non-owner and Non-Administrator users should not be allowed to stop room recordings.""" """Non-owner and Non-Administrator users should not be allowed to stop room recordings."""
settings.RECORDING_ENABLE = True
room = RoomFactory() room = RoomFactory()
user = UserFactory() user = UserFactory()
RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE) RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
@@ -84,8 +85,8 @@ def test_stop_recording_recording_disabled(settings):
response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/") response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
assert response.status_code == 403 assert response.status_code == 404
assert response.json() == {"detail": "Access denied, recording is disabled."} assert response.json() == {"detail": "Not found."}
# Verify no recording exists # Verify no recording exists
assert Recording.objects.count() == 0 assert Recording.objects.count() == 0

View File

@@ -128,8 +128,8 @@ def test_start_subtitle_disabled_by_default(mock_livekit_token):
{"token": mock_livekit_token}, {"token": mock_livekit_token},
) )
assert response.status_code == 403 assert response.status_code == 404
assert response.json() == {"detail": "Access denied, subtitles are disabled."} assert response.json() == {"detail": "Not found."}
def test_start_subtitle_valid_token( def test_start_subtitle_valid_token(