mirror of
https://github.com/suitenumerique/meet.git
synced 2026-10-12 07:57:50 +00:00
Lebaud Antoine
233 lines
7.5 KiB
Python
233 lines
7.5 KiB
Python
"""Permission handlers for the Meet core app."""
|
|
|
|
from django.conf import settings
|
|
from django.http import Http404
|
|
|
|
from rest_framework import permissions
|
|
|
|
from ..models import RoleChoices, RoomAccessLevel
|
|
from ..services.participants_management import (
|
|
ParticipantNotFoundException,
|
|
ParticipantsManagement,
|
|
ParticipantsManagementException,
|
|
)
|
|
|
|
|
|
class IsAuthenticated(permissions.BasePermission):
|
|
"""
|
|
Allows access only to authenticated users. Alternative method checking the presence
|
|
of the auth token to avoid hitting the database.
|
|
"""
|
|
|
|
def has_permission(self, request, view):
|
|
return bool(request.auth) or request.user.is_authenticated
|
|
|
|
|
|
class IsSelf(IsAuthenticated):
|
|
"""
|
|
Allows access only to authenticated users. Alternative method checking the presence
|
|
of the auth token to avoid hitting the database.
|
|
"""
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""Write permissions are only allowed to the user itself."""
|
|
return obj == request.user
|
|
|
|
|
|
class RoomPermissions(permissions.BasePermission):
|
|
"""
|
|
Permissions applying to the room API endpoint.
|
|
"""
|
|
|
|
def has_permission(self, request, view):
|
|
"""Only allow authenticated users for unsafe methods."""
|
|
if request.method in permissions.SAFE_METHODS:
|
|
return True
|
|
|
|
return request.user.is_authenticated
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""Object permissions are only given to administrators of the room."""
|
|
|
|
if request.method in permissions.SAFE_METHODS:
|
|
return True
|
|
|
|
user = request.user
|
|
|
|
if request.method == "DELETE":
|
|
return obj.is_owner(user)
|
|
|
|
return obj.is_administrator_or_owner(user)
|
|
|
|
|
|
class ResourceAccessPermission(IsAuthenticated):
|
|
"""
|
|
Permissions for a room that can only be updated by room administrators.
|
|
"""
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""
|
|
Check that the logged-in user is administrator of the linked room.
|
|
"""
|
|
user = request.user
|
|
if request.method == "DELETE" and obj.role == RoleChoices.OWNER:
|
|
return obj.user == user
|
|
|
|
return obj.resource.is_administrator_or_owner(user)
|
|
|
|
|
|
class HasAbilityPermission(IsAuthenticated):
|
|
"""Permission class for access objects."""
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""Check permission for a given object."""
|
|
return obj.get_abilities(request.user).get(view.action, False)
|
|
|
|
|
|
class HasPrivilegesOnRoom(IsAuthenticated):
|
|
"""Check if user has privileges on a given room."""
|
|
|
|
message = "You must have privileges on room to perform this action."
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""Determine if user has privileges on room."""
|
|
return obj.is_administrator_or_owner(request.user)
|
|
|
|
|
|
class HasLiveKitRoomAccess(permissions.BasePermission):
|
|
"""Check if authenticated user's LiveKit token is for the specific room."""
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
if not request.auth or not hasattr(request.auth, "video"):
|
|
return False
|
|
return request.auth.video.room == str(obj.id)
|
|
|
|
|
|
class FilePermission(IsAuthenticated):
|
|
"""
|
|
Permissions applying to the file API endpoint.
|
|
Handling soft deletions specificities
|
|
"""
|
|
|
|
def has_permission(self, request, view):
|
|
"""Allow access only to authenticated users."""
|
|
if not settings.FILE_UPLOAD_ENABLED:
|
|
raise Http404
|
|
|
|
return super().has_permission(request, view)
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""
|
|
Return a 404 on deleted files or if the user is not the owner
|
|
"""
|
|
|
|
if obj.deleted_at is not None or obj.hard_deleted_at is not None:
|
|
raise Http404
|
|
|
|
if obj.creator != request.user:
|
|
raise Http404
|
|
|
|
return obj.get_abilities(request.user).get(view.action, False)
|
|
|
|
|
|
class CanMuteParticipant(permissions.BasePermission):
|
|
"""
|
|
Grant muting rights based on role or room configuration.
|
|
|
|
- Admins and owners can always mute.
|
|
- When `everyone_can_mute` is enabled on the room, any participant
|
|
currently in the room (proven by a valid LiveKit token for that room)
|
|
can mute.
|
|
"""
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""Check if the requesting user is allowed to mute a participant in the given room."""
|
|
|
|
is_livekit_token_auth = request.auth and hasattr(request.auth, "video")
|
|
|
|
# Always allow admins/owners when authenticated with session cookie
|
|
if not is_livekit_token_auth and obj.is_administrator_or_owner(request.user):
|
|
return True
|
|
|
|
everyone_can_mute = obj.configuration.get("everyone_can_mute", True)
|
|
if not everyone_can_mute:
|
|
return False
|
|
|
|
if not is_livekit_token_auth:
|
|
return False
|
|
|
|
# LiveKit token scoped to this room
|
|
return request.auth.video.room == str(obj.id)
|
|
|
|
|
|
class IsPresentInMeeting(permissions.BasePermission):
|
|
"""Check that the requesting user is currently connected to the meeting.
|
|
|
|
The requester must be session-authenticated (their DB identity is needed
|
|
to check privileges); presence is verified against LiveKit using their
|
|
`sub` as participant identity. Fails closed on LiveKit errors.
|
|
"""
|
|
|
|
message = "You must be connected to the meeting to perform this action."
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
"""Verify the requester's identity is a participant of the room."""
|
|
user = request.user
|
|
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
|
|
try:
|
|
return ParticipantsManagement().check_if_in_meeting(
|
|
room_name=str(obj.pk), identity=str(user.sub)
|
|
)
|
|
except ParticipantNotFoundException:
|
|
return False
|
|
except ParticipantsManagementException:
|
|
return False
|
|
|
|
|
|
class CanManageLobby(permissions.BasePermission):
|
|
"""Grant lobby management (list/accept/deny waiting participants).
|
|
|
|
- Room admins/owners can always manage the lobby.
|
|
- When the room access level is TRUSTED, any authenticated user who is
|
|
currently connected to the meeting can manage the lobby. Presence is
|
|
verified cache-first (Redis), falling back to the LiveKit API.
|
|
|
|
Access level is always read fresh from the DB; only presence is cached,
|
|
so changing the room to RESTRICTED takes effect immediately.
|
|
"""
|
|
|
|
message = "You are not allowed to manage this room's lobby."
|
|
|
|
# pylint: disable=too-many-return-statements
|
|
def has_object_permission(self, request, view, obj): # noqa: PLR0911
|
|
"""Check privileges first, then the trusted-room presence path."""
|
|
user = request.user
|
|
|
|
if not user or not user.is_authenticated:
|
|
return False
|
|
|
|
# Product choice: lobby management is reserved for session-authenticated
|
|
# users with a real account, not holders of a LiveKit room token.
|
|
if request.auth and hasattr(request.auth, "video"):
|
|
return False
|
|
|
|
if obj.is_administrator_or_owner(user):
|
|
return True
|
|
|
|
if obj.access_level != RoomAccessLevel.TRUSTED:
|
|
return False
|
|
|
|
self.message = "You must be connected to the meeting to manage its lobby."
|
|
|
|
try:
|
|
return ParticipantsManagement().check_if_in_meeting_cached(
|
|
room_name=str(obj.pk), identity=str(user.sub)
|
|
)
|
|
except ParticipantNotFoundException:
|
|
return False
|
|
except ParticipantsManagementException:
|
|
return False
|