meet/src/backend/core/utils.py

"""
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 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(
    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,
) -> 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.

    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",
            }
        )
    )

    return token.to_jwt()


def generate_livekit_config(
    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 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

    # This settings should be used if the backend application and the frontend application
    # can't connect to the object storage with the same domain. This is the case in the
    # docker compose stack used in development. The frontend application will use localhost
    # to connect to the object storage while the backend application will use the object storage
    # service name declared in the docker compose stack.
    # This is needed because the domain name is used to compute the signature. So it can't be
    # changed dynamically by the frontend application.
    if settings.AWS_S3_DOMAIN_REPLACE:
        s3_client = 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,
            ),
        )
    else:
        s3_client = default_storage.connection.meta.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")

    # This setting should be used if the backend application and the frontend application
    # can't connect to the object storage with the same domain. This is the case in the
    # docker compose stack used in development. The frontend application will use localhost
    # to connect to the object storage while the backend application will use the object storage
    # service name declared in the docker compose stack.
    # This is needed because the domain name is used to compute the signature. So it can't be
    # changed dynamically by the frontend application.
    if settings.AWS_S3_DOMAIN_REPLACE and override_domain:
        s3_client = 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,
            ),
        )
    else:
        s3_client = default_storage.connection.meta.client

    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,
    }