""" Utils functions used in the core app """ # pylint: disable=R0913, R0917 # ruff: noqa:S311, PLR0913 import hashlib import json import logging import mimetypes import random import secrets import string from datetime import timedelta from functools import lru_cache from typing import List, Optional from uuid import uuid4 from django.conf import settings from django.core.files.storage import default_storage import aiohttp import boto3 import botocore import magic import phonenumbers from asgiref.sync import async_to_sync from livekit.api import ( # pylint: disable=E0611 AccessToken, ListRoomsRequest, LiveKitAPI, SendDataRequest, TwirpError, VideoGrants, ) logger = logging.getLogger(__name__) def generate_color(identity: str) -> str: """Generates a consistent HSL color based on a given identity string. The function seeds the random generator with the identity's hash, ensuring consistent color output. The HSL format allows fine-tuned control over saturation and lightness, empirically adjusted to produce visually appealing and distinct colors. HSL is preferred over hex to constrain the color range and ensure predictability. """ # ruff: noqa:S324 identity_hash = hashlib.sha1(identity.encode("utf-8")) # Keep only hash's last 16 bits, collisions are not a concern seed = int(identity_hash.hexdigest(), 16) & 0xFFFF random.seed(seed) hue = random.randint(0, 360) saturation = random.randint(50, 75) lightness = random.randint(25, 60) return f"hsl({hue}, {saturation}%, {lightness}%)" def generate_token( # noqa: PLR0917 room: str, user, username: Optional[str] = None, color: Optional[str] = None, sources: Optional[List[str]] = None, role: Optional[str] = None, participant_id: Optional[str] = None, ttl: Optional[timedelta] = None, ) -> str: """Generate a LiveKit access token for a user in a specific room. Args: room (str): The name of the room. user (User): The user which request the access token. username (Optional[str]): The username to be displayed in the room. If none, a default value will be used. color (Optional[str]): The color to be displayed in the room. If none, a value will be generated sources: (Optional[List[str]]): List of media sources the user can publish If none, defaults to LIVEKIT_DEFAULT_SOURCES. role (Optional[str]): Room's access role if any participant_id (Optional[str]): Stable identifier for anonymous users; used as identity when user.is_anonymous. ttl (Optional[timedelta]): Token validity duration. Defaults to LiveKit SDK default. Returns: str: The LiveKit JWT access token. """ is_admin_or_owner = role in ("owner", "administrator") if is_admin_or_owner: sources = settings.LIVEKIT_DEFAULT_SOURCES if sources is None: sources = settings.LIVEKIT_DEFAULT_SOURCES video_grants = VideoGrants( room=room, room_join=True, room_admin=is_admin_or_owner, can_update_own_metadata=False, can_publish=bool(sources), can_publish_sources=sources, can_subscribe=True, ) if user.is_anonymous: identity = participant_id or str(uuid4()) default_username = "Anonymous" else: identity = str(user.sub) default_username = user.full_name or str(user) if color is None: color = generate_color(identity) can_edit = ( settings.AUTHENTICATED_PARTICIPANTS_CAN_EDIT_DISPLAY_NAME or user.is_anonymous ) display_name = (username or default_username) if can_edit else default_username token = ( AccessToken( api_key=settings.LIVEKIT_CONFIGURATION["api_key"], api_secret=settings.LIVEKIT_CONFIGURATION["api_secret"], ) .with_grants(video_grants) .with_identity(identity) .with_name(display_name) .with_attributes( { "color": color, "room_role": role, "is_authenticated": "true" if user.is_authenticated else "false", } ) ) if ttl is not None: token = token.with_ttl(ttl) return token.to_jwt() def generate_livekit_config( # noqa: PLR0917 room_id: str, user, username: str, role: Optional[str] = None, color: Optional[str] = None, configuration: Optional[dict] = None, participant_id: Optional[str] = None, ) -> dict: """Generate LiveKit configuration for room access. Args: room_id: Room identifier user: User instance requesting access username: Display name in room role (str): Room's access role if any color (Optional[str]): Optional color to associate with the participant. configuration (Optional[dict]): Room configuration dict that can override default settings. participant_id (Optional[str]): Stable identifier for anonymous users; used as identity when user.is_anonymous. Returns: dict: LiveKit configuration with URL, room and access token """ sources = None if configuration is not None: sources = configuration.get("can_publish_sources", None) return { "url": settings.LIVEKIT_CONFIGURATION["url"], "room": room_id, "token": generate_token( room=room_id, user=user, username=username, color=color, sources=sources, role=role, participant_id=participant_id, ), } def generate_s3_authorization_headers(key): """ Generate authorization headers for an s3 object. These headers can be used as an alternative to signed urls with many benefits: - the urls of our files never expire and can be stored in our recording' metadata - we don't leak authorized urls that could be shared (file access can only be done with cookies) - access control is truly realtime - the object storage service does not need to be exposed on internet """ url = default_storage.unsigned_connection.meta.client.generate_presigned_url( "get_object", ExpiresIn=0, Params={"Bucket": default_storage.bucket_name, "Key": key}, ) request = botocore.awsrequest.AWSRequest(method="get", url=url) s3_client = default_storage.connection.meta.client # pylint: disable=protected-access credentials = s3_client._request_signer._credentials # noqa: SLF001 frozen_credentials = credentials.get_frozen_credentials() region = s3_client.meta.region_name auth = botocore.auth.S3SigV4Auth(frozen_credentials, "s3", region) auth.add_auth(request) return request def create_livekit_client(custom_configuration=None): """Create and return a configured LiveKit API client.""" custom_session = None if not settings.LIVEKIT_VERIFY_SSL: connector = aiohttp.TCPConnector(ssl=False) custom_session = aiohttp.ClientSession(connector=connector) # Use default configuration if none provided configuration = custom_configuration or settings.LIVEKIT_CONFIGURATION return LiveKitAPI(session=custom_session, **configuration) class NotificationError(Exception): """Notification delivery to room participants fails.""" @async_to_sync async def notify_participants(room_name: str, notification_data: dict): """Send notification data to all participants in a LiveKit room.""" lkapi = create_livekit_client() try: room_response = await lkapi.room.list_rooms( ListRoomsRequest( names=[room_name], ) ) # Check if the room exists if not room_response.rooms: return await lkapi.room.send_data( SendDataRequest( room=room_name, data=json.dumps(notification_data).encode("utf-8"), kind="RELIABLE", ) ) except TwirpError as e: raise NotificationError("Failed to notify room participants") from e finally: await lkapi.aclose() ALPHANUMERIC_CHARSET = string.ascii_letters + string.digits def generate_secure_token(length: int = 30, charset: str = ALPHANUMERIC_CHARSET) -> str: """Generate a cryptographically secure random token. Uses SystemRandom for proper entropy, suitable for OAuth tokens and API credentials that must be non-guessable. Inspired by: https://github.com/oauthlib/oauthlib/blob/master/oauthlib/common.py Args: length: Token length in characters (default: 30) charset: Character set to use for generation Returns: Cryptographically secure random token """ return "".join(secrets.choice(charset) for _ in range(length)) def generate_client_id() -> str: """Generate a unique client ID for application authentication. Returns: Random client ID string """ return generate_secure_token(settings.APPLICATION_CLIENT_ID_LENGTH) def generate_client_secret() -> str: """Generate a secure client secret for application authentication. Returns: Cryptographically secure client secret """ return generate_secure_token(settings.APPLICATION_CLIENT_SECRET_LENGTH) def generate_room_slug(): """Generate a random room slug in the format 'xxx-xxxx-xxx'.""" sizes = [3, 4, 3] parts = [ "".join(secrets.choice(string.ascii_lowercase) for _ in range(size)) for size in sizes ] return "-".join(parts) def detect_mimetype(file_buffer: bytes, filename: str | None = None) -> str: """ Detect MIME type using multiple methods for better accuracy. This function combines: 1. Magic bytes detection (python-magic) - most reliable for actual file content 2. File extension detection (mimetypes) - useful as fallback or for validation Args: file_buffer: The file content buffer (first bytes of the file) filename: Optional filename to extract extension from Returns: str: The detected MIME type Notes: Originally from https://github.com/suitenumerique/drive/blob/564822d31f071c6dfacd112ef4b7146c73077cd9/src/backend/core/api/utils.py#L166 # pylint:disable=line-too-long """ # Initialize magic detector mime_detector = magic.Magic(mime=True) # Method 1: Detect from file content (magic bytes) - most reliable mimetype_from_content = mime_detector.from_buffer(file_buffer) # If we have a filename, try extension-based detection as well mimetype_from_extension = None if filename: # Use mimetypes module to guess from extension # Use guess_file_type (Python 3.13+) instead of deprecated guess_type mimetype_from_extension, _ = mimetypes.guess_file_type(filename, strict=False) logger.debug("detect_mimetype: mimetype_from_content: %s", mimetype_from_content) logger.debug( "detect_mimetype: mimetype_from_extension: %s", mimetype_from_extension ) # Strategy: Prefer content-based detection, but use extension if: # 1. Content detection returns generic types (application/octet-stream, text/plain) # 2. Content detection fails or returns None # 3. Extension detection provides a more specific type # Generic/unreliable MIME types that we should try to improve generic_types = { "application/octet-stream", "application/x-ole-storage", # used by .xls, .doc and .ppt "application/zip", "text/plain", } # If content detection gives us a generic type and we have extension info if mimetype_from_content in generic_types and mimetype_from_extension: # Use extension-based detection if it's more specific if mimetype_from_extension not in generic_types: return mimetype_from_extension # If content detection failed, returned None or is a generic type, use extension if available if not mimetype_from_content or mimetype_from_content in generic_types: if mimetype_from_extension: return mimetype_from_extension # Default to content-based detection (most reliable) return mimetype_from_content or "application/octet-stream" def _get_s3_client(*, override_domain: bool = True): """Return an S3 client, honoring the AWS_S3_DOMAIN_REPLACE endpoint override. AWS_S3_DOMAIN_REPLACE is used when the backend and frontend reach object storage under different domains (this is the case in the docker compose stack used in development: the frontend connects to the object storage on localhost while the backend uses the object storage service name declared in the stack). The domain name is used to compute the signature, so it can't be changed dynamically by the frontend; we build a dedicated boto3 client pointed at that endpoint. Otherwise we reuse the default storage client. """ if settings.AWS_S3_DOMAIN_REPLACE and override_domain: return boto3.client( "s3", aws_access_key_id=settings.AWS_S3_ACCESS_KEY_ID, aws_secret_access_key=settings.AWS_S3_SECRET_ACCESS_KEY, endpoint_url=settings.AWS_S3_DOMAIN_REPLACE, config=botocore.client.Config( region_name=settings.AWS_S3_REGION_NAME, signature_version=settings.AWS_S3_SIGNATURE_VERSION, ), ) return default_storage.connection.meta.client def generate_upload_policy(file): """ Generate a S3 upload policy for a given file. Notes: Originally taken from https://github.com/suitenumerique/drive/blob/564822d31f071c6dfacd112ef4b7146c73077cd9/src/backend/core/api/utils.py#L102 # pylint: disable=line-too-long """ key = file.temporary_file_key s3_client = _get_s3_client() # Generate the policy policy = s3_client.generate_presigned_url( ClientMethod="put_object", Params={"Bucket": default_storage.bucket_name, "Key": key, "ACL": "private"}, ExpiresIn=settings.AWS_S3_UPLOAD_POLICY_EXPIRATION, ) return policy def generate_download_s3_url( key: str, *, expires_in: int, override_domain: bool = True ): """ Generate a S3 signed download url for a given key. """ if not key: raise ValueError("key cannot be empty") s3_client = _get_s3_client(override_domain=override_domain) return s3_client.generate_presigned_url( ClientMethod="get_object", Params={"Bucket": default_storage.bucket_name, "Key": key}, ExpiresIn=expires_in, ) @lru_cache(maxsize=1) def _format_telephony_phone_number(raw_number, default_country): """Parse a configured phone number and return (country, international_format). Returns (None, None) if the inputs are missing or the number cannot be parsed. Logs a warning on parse failure so operators see the misconfiguration. """ if not raw_number or not default_country: return None, None try: parsed = phonenumbers.parse(raw_number, default_country) except phonenumbers.NumberParseException: logger.warning( "ROOM_TELEPHONY_PHONE_NUMBER %r is not a valid phone number for " "default country %r; telephony block will be returned without " "formatted number.", raw_number, default_country, ) return None, None country = phonenumbers.region_code_for_number(parsed) international = phonenumbers.format_number( parsed, phonenumbers.PhoneNumberFormat.INTERNATIONAL ) return country, international def build_telephony_config(): """Build the telephony block of the frontend configuration.""" if not settings.ROOM_TELEPHONY_ENABLED: return {"enabled": False} country, international = _format_telephony_phone_number( settings.ROOM_TELEPHONY_PHONE_NUMBER, settings.ROOM_TELEPHONY_DEFAULT_COUNTRY, ) if international is None: logger.warning( "Telephony is enabled but ROOM_TELEPHONY_PHONE_NUMBER %r with " "default country %r could not be formatted; telephony will be disabled.", settings.ROOM_TELEPHONY_PHONE_NUMBER, settings.ROOM_TELEPHONY_DEFAULT_COUNTRY, ) return {"enabled": False} return { "enabled": True, "default_country": country, "international_phone_number": international, } CACHE_SCAN_ITERSIZE = 500