romm/backend/handler/streaming/access.py

"""Who may see and drive a streaming session, and which container holds it."""

from typing import Any

from fastapi import HTTPException, Request

from handler.auth.dependencies import get_permissions
from handler.database import db_platform_handler, db_rom_handler
from handler.streaming.config import (
    ResolvedContainer,
    containers_by_key,
    containers_for_platform,
    entry_for_platform,
    key_for_name,
)
from handler.streaming.session_store import (
    get_live_session,
    session_is_desktop,
    session_platform_matches,
)
from logger.logger import log
from models.rom import Rom
from models.user import Role


def session_owner_id(session: dict[str, Any], request: Request) -> int:
    """Whose library a session's states and saves belong to.

    An admin may drive a session they do not own, and what it writes is still
    the player's, so the acting user only stands in for a record that names
    nobody.
    """
    user_id = session.get("user_id")
    return user_id if isinstance(user_id, int) else request.user.id


def assert_session_owner(session: dict[str, Any], request: Request) -> None:
    """Only the user who claimed a session (or an admin) may control it."""
    if session.get("user_id") == request.user.id:
        return
    if request.user.role == Role.ADMIN:
        return
    raise HTTPException(status_code=403, detail="Session is claimed by another user")


def rom_is_visible(request: Request, rom: Rom | None) -> bool:
    """Can the caller see this ROM?

    Sessions outlive nothing but the cache, so a rom_id that no longer
    resolves is treated as visible: there is no hidden ROM left to protect.
    """
    if rom is None:
        return True
    if not request.user.is_authenticated:
        return True
    return get_permissions(request).can_see_rom(rom.id, rom.platform_id)


def session_rom(session: dict[str, Any]) -> Rom | None:
    """The ROM a session is running, None for a desktop or a deleted entry."""
    rom_id = session.get("rom_id")
    return db_rom_handler.get_rom_simple(rom_id) if rom_id is not None else None


def session_rom_is_visible(request: Request, session: dict[str, Any]) -> bool:
    """Can the caller see the ROM a session is running?"""
    return rom_is_visible(request, session_rom(session))


def joinable_session_rom(
    request: Request, session: dict[str, Any]
) -> tuple[bool, Rom | None]:
    """Whether a live session is one this caller may ask to join (its host opted
    into multiplayer, it is somebody else's, its ROM is not hidden), with that ROM."""
    if not session.get("multiplayer") or session.get("user_id") == request.user.id:
        return False, None
    rom = session_rom(session)
    return rom_is_visible(request, rom), rom


def assert_session_rom_visible(
    request: Request, session: dict[str, Any], *, not_found_detail: str
) -> None:
    """Raise 404 when the session's ROM is hidden from the caller."""
    if not session_rom_is_visible(request, session):
        raise HTTPException(status_code=404, detail=not_found_detail)


def visible_rom_name(request: Request, session: dict[str, Any]) -> str | None:
    """The name of the ROM a session is running, blanked when the caller cannot
    see that ROM. "Busy" is safe to report to anyone; what is running is not."""
    if not session or not session_rom_is_visible(request, session):
        return None
    name = session.get("rom_name")
    return str(name) if name else None


def platform_is_visible(request: Request, platform_slug: str) -> bool:
    """Can the caller see this platform?

    A container may name a slug the library has never scanned, which has no
    platform row and so nothing to hide.
    """
    if not platform_slug or not request.user.is_authenticated:
        return True
    platform = db_platform_handler.get_platform_by_slug(platform_slug)
    if platform is None:
        return True
    return get_permissions(request).can_see_platform(platform.id)


def session_in_scope(
    session: dict[str, Any],
    platform: str,
    include_desktop: bool,
    claimed_at: str | None = None,
) -> bool:
    """Whether a stored session is the one a route asked about, which its owner
    alone cannot say: same platform, a desktop only if named, the given claim."""
    if not include_desktop and session_is_desktop(session):
        return False
    if claimed_at is not None and session.get("claimed_at") != claimed_at:
        return False
    return session_platform_matches(session, platform)


def notice_in_scope(
    notice: dict[str, Any],
    platform: str,
    include_desktop: bool,
    claimed_at: str | None = None,
) -> bool:
    """Whether a notice answers for the claim a route asked about: it records the
    ended claim's platform, kind and stamp, so a session's scope holds for its
    tombstone."""
    return session_in_scope(notice, platform, include_desktop, claimed_at)


async def find_session_for_user(
    candidates: list[ResolvedContainer],
    user_id: int,
    *,
    platform: str,
    include_desktop: bool = False,
    claimed_at: str | None = None,
) -> tuple[ResolvedContainer, str, dict[str, Any]] | None:
    """The candidate holding this user's session, as (container, key, session).

    With a pool the platform no longer identifies the container, the session does.

    Args:
        platform: only sessions claimed for it match.
        include_desktop: whether a desktop matches, only when its container is named.
        claimed_at: only the one claim the caller was given matches.
    """
    for candidate in candidates:
        session_key = candidate.key
        session = await get_live_session(session_key)
        if session is None:
            continue
        if not session_in_scope(session, platform, include_desktop, claimed_at):
            continue
        if session.get("user_id") == user_id:
            return candidate, session_key, session
    return None


def named_container(platform: str, container_key: str) -> ResolvedContainer:
    """A container serving a platform, found by key so one in a later pool stays
    reachable. Raises 404 when the key names no such container."""
    candidate = entry_for_platform(containers_by_key().get(container_key, []), platform)
    if candidate is None:
        raise HTTPException(
            status_code=404,
            detail=f"No streaming container '{container_key}' for platform '{platform}'",
        )
    return candidate


async def resolve_named_container(
    platform: str, container_key: str
) -> tuple[ResolvedContainer, str, dict[str, Any] | None]:
    """A named container plus whatever session it holds.

    Returns (container, session_key, session), the session being None when the
    container is free or draining.
    """
    return (
        named_container(platform, container_key),
        container_key,
        await get_live_session(container_key),
    )


async def resolve_owned_session(
    platform: str, request: Request
) -> tuple[ResolvedContainer, str, dict[str, Any]]:
    """Find the caller's session for this platform in its first pool.

    Returns (container, session_key, session). Raises 404 when the platform has
    no configured container or nothing is active, 403 when every active session
    belongs to someone else, 409 when an admin's fallback is ambiguous.
    """
    candidates = containers_for_platform(platform)
    if not candidates:
        raise HTTPException(
            status_code=404,
            detail=f"No streaming container configured for platform '{platform}'",
        )

    others: list[tuple[ResolvedContainer, str, dict[str, Any]]] = []
    for candidate in candidates:
        session_key = candidate.key
        session = await get_live_session(session_key)
        if session is None:
            continue
        # These routes act on a game; a desktop is reached by naming it.
        if not session_in_scope(session, platform, include_desktop=False):
            continue
        if session.get("user_id") == request.user.id:
            return candidate, session_key, session
        others.append((candidate, session_key, session))

    if not others:
        raise HTTPException(
            status_code=404, detail=f"No active session for platform '{platform}'"
        )
    # An admin may control a session they do not own, but the scan found none of
    # theirs, so fall back to the platform's active session. A pool can hold
    # several and the path does not say which, so the caller has to name one.
    if request.user.role != Role.ADMIN:
        raise HTTPException(
            status_code=403, detail="Session is claimed by another user"
        )
    if len(others) > 1:
        raise HTTPException(
            status_code=409,
            detail=(
                f"{len(others)} sessions are active on platform '{platform}', "
                "name a container instead"
            ),
        )
    return others[0]


async def resolve_claim(
    platform: str,
    request: Request,
    container_key: str | None,
    claimed_at: str | None,
    *,
    game_only: bool,
) -> tuple[ResolvedContainer, str, dict[str, Any]] | None:
    """The session a route acting on a claim reaches, or None when that claim is gone.

    Args:
        container_key: the container the caller named, if any.
        claimed_at: the stamp of the one claim the caller was given.
        game_only: only a game of this platform matches a named container, while
            a release ends whatever it holds.
    """
    if container_key is not None:
        container, session_key, session = await resolve_named_container(
            platform, container_key
        )
        if session is None:
            return None
        if game_only and not session_in_scope(session, platform, include_desktop=False):
            return None
    else:
        try:
            container, session_key, session = await resolve_owned_session(
                platform, request
            )
        except HTTPException as exc:
            # Nothing of the caller's is active, so a stamped claim is gone and a
            # repeated call from the same tab finds nothing: both are a no-op.
            if exc.status_code != 404 and claimed_at is None:
                raise
            return None

    # A tab whose claim was replaced (a takeover, or Play pressed again elsewhere)
    # must not act on the claim that took its place.
    if claimed_at is not None and session.get("claimed_at") != claimed_at:
        log.info("%s ignored for a replaced claim", request.url.path)
        return None
    # After the stamp check, so another player's takeover reads as a gone claim.
    if container_key is not None:
        assert_session_owner(session, request)
    return container, session_key, session


async def require_claim(
    platform: str,
    request: Request,
    container_key: str | None,
    claimed_at: str | None,
) -> tuple[ResolvedContainer, str, dict[str, Any]]:
    """The game a control route drives, raising 404 once the caller's claim is gone."""
    target = await resolve_claim(
        platform, request, container_key, claimed_at, game_only=True
    )
    if target is None:
        raise HTTPException(
            status_code=404, detail=f"No active session for platform '{platform}'"
        )
    return target


def container_by_name(name: str) -> tuple[ResolvedContainer, str]:
    """A configured container named by its name or key, plus the platform to
    file its sessions under.

    The session routes are platform-keyed, so a container serving several gets
    the first, which `container_for_session` resolves back to this entry.
    Raises 404 when the name matches no container.
    """
    key = key_for_name(name) if name else None
    entries = containers_by_key().get(key) if key else None
    if not entries:
        raise HTTPException(status_code=404, detail=f"No streaming container '{name}'")
    return entries[0], entries[0].platform