mirror of
https://github.com/rommapp/romm.git
synced 2026-10-12 02:21:58 +00:00
Zach Clendenen
4f2d1198f2
Several notes had run to four or more lines, a couple of them recounting how the code was reordered rather than what it does now.
255 lines
9.3 KiB
Python
255 lines
9.3 KiB
Python
"""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,
|
|
)
|
|
from handler.streaming.session_store import (
|
|
get_live_session,
|
|
session_is_desktop,
|
|
session_platform_matches,
|
|
)
|
|
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 session_is_joinable(
|
|
request: Request, session: dict[str, Any], rom: Rom | None
|
|
) -> bool:
|
|
"""Whether a live session is one this caller may ask to join: its host opted
|
|
into multiplayer, it is somebody else's, and its ROM is not hidden."""
|
|
return (
|
|
bool(session.get("multiplayer"))
|
|
and session.get("user_id") != request.user.id
|
|
and rom_is_visible(request, 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 | None,
|
|
include_desktop: bool,
|
|
claimed_at: str | None = None,
|
|
) -> bool:
|
|
"""Whether a stored session is the one the route asked about: the owner alone
|
|
does not say, since a container holds one session across several platforms,
|
|
a desktop answers only to a caller that named it, and a claim is its holder
|
|
plus the moment they took it."""
|
|
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 platform is None or session_platform_matches(session, platform)
|
|
|
|
|
|
def notice_in_scope(
|
|
notice: dict[str, Any], platform: str, include_desktop: bool
|
|
) -> bool:
|
|
"""Whether a notice answers for the claim a route asked about: it records the
|
|
ended claim's platform and kind, so a session's scope holds for its tombstone."""
|
|
return _session_in_scope(notice, platform, include_desktop)
|
|
|
|
|
|
async def find_session_for_user(
|
|
candidates: list[ResolvedContainer],
|
|
user_id: int,
|
|
*,
|
|
platform: str | None = None,
|
|
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. `platform` scopes the match to sessions claimed for it, `claimed_at`
|
|
to the one claim the caller was given, and a desktop is only reachable when
|
|
the caller named its container.
|
|
"""
|
|
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]
|
|
|
|
|
|
def container_by_key(container_key: str) -> tuple[ResolvedContainer, str]:
|
|
"""A configured container named by its 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 key names no container.
|
|
"""
|
|
entries = containers_by_key().get(container_key) if container_key else None
|
|
if not entries:
|
|
raise HTTPException(
|
|
status_code=404, detail=f"No streaming container '{container_key}'"
|
|
)
|
|
return entries[0], entries[0].platform
|