WebRTCPeerManagement

def __init__(self, options: Any) -> None:
    self.peers: Dict[str, Peer] = {}
    self.sessions: Dict[str, str] = {}
    self.rooms: Dict[str, Set[str]] = {}
    self.on_client_presence: Optional[Callable[[bool], None]] = None
    self._eviction_times: Dict[Any, List[float]] = {}
    self._holds: Dict[str, "asyncio.Task[None]"] = {}

    self.keepalive_timeout: int = options.keepalive_timeout

    self.turn_shared_secret: Optional[str] = options.turn_shared_secret
    self.turn_host: Optional[str] = options.turn_host
    self.turn_port: Optional[int] = options.turn_port
    self.turn_protocol: str = options.turn_protocol.lower()
    if self.turn_protocol != "tcp":
        self.turn_protocol = "udp"
    self.turn_tls: Optional[bool] = options.turn_tls
    self.turn_auth_header_name: str = options.turn_auth_header_name
    self.stun_host: Optional[str] = options.stun_host
    self.stun_port: Optional[int] = options.stun_port

    self.enable_sharing: bool = options.enable_sharing
    self.enable_shared: bool = options.enable_shared
    self.enable_player2: bool = options.enable_player2
    self.enable_player3: bool = options.enable_player3
    self.enable_player4: bool = options.enable_player4
    self.lock: asyncio.Lock = asyncio.Lock()

    self.rtc_config: Optional[Any] = getattr(options, "rtc_config", None)
    if not self.rtc_config and os.path.exists(options.rtc_config_file):
        logger.debug("parsing rtc_config_file: {}".format(options.rtc_config_file))
        if _is_trusted_config_file(options.rtc_config_file):
            self.rtc_config = self.read_file(options.rtc_config_file)
        else:
            logger.error(
                "Refusing to use RTC config file {!r}: unsafe ownership or permissions.".format(
                    options.rtc_config_file
                )
            )
    if self.turn_shared_secret:
        if not (self.turn_host and self.turn_port):
            raise Exception(
                "missing turn_host or turn_port options with turn_shared_secret"
            )

def read_file(self, path: str) -> Optional[str]:
    """Read and return the contents of a file as a string.

    Args:
        path: Path to the file to read.

    Returns:
        File contents as a string, or None if the file cannot be read.
    """
    try:
        with open(path, "r") as f:
            return f.read()
    except OSError as e:
        logger.error("Failed to read rtc_config_file {!r}: {}".format(path, e))
        return None

def set_rtc_config(self, rtc_config: Any) -> None:
    """Set the RTC configuration served to clients.

    Args:
        rtc_config: RTC configuration as a JSON string or dict.
    """
    self.rtc_config = rtc_config

async def recv_msg_ping(self, ws: WebSocketResponse, raddr: str) -> WSMessage:
    """Wait for a message, sending periodic pings to prevent connection timeout.

    Args:
        ws: The peer's WebSocket.
        raddr: Remote address, for logging.

    Returns:
        The received message; blocks (with keepalive pings on each
        ``keepalive_timeout``) until one arrives.

    Raises:
        Exception: The socket closed while waiting.
    """
    msg_obj: Optional[Any] = None
    while msg_obj is None:
        try:
            if ws.closed:
                raise Exception(f"Websocket Connection closed with {ws.close_code} status code")
            msg_obj = await asyncio.wait_for(ws.receive(), self.keepalive_timeout)
        except (asyncio.TimeoutError, concurrent.futures.TimeoutError):
            logger.debug("Sending keepalive ping to {!r} in recv".format(raddr))
            await ws.ping()
        except (ConnectionResetError, ConnectionError, Exception):
            raise
    return msg_obj

async def cleanup_session(self, uid: str) -> None:
    """Clean up a session when a peer disconnects.

    Every client — a primary controller included — only tells the server
    its session ended: the server socket is shared by every display's
    session and every viewer, so dropping it would reset the whole
    session. The server side reaps just this peer and keeps a reconnect
    grace on the primary capture, so a controller tab reload never kicks
    the viewers or the second display (websockets `_teardown_if_unclaimed`
    parity).

    Args:
        uid: Peer ID to clean up.
    """
    if uid in self.sessions:
        other_id = self.sessions[uid]
        del self.sessions[uid]
        peer = self.peers.get(uid)
        client_type = peer.client_type if peer else "unknown"
        logger.debug(
            "Cleaned up {} session, client type {!r}".format(
                uid, client_type
            )
        )

        if not peer:
            return
        if peer.client_type in ("controller", "viewer"):
            other_peer = self.peers.get(other_id)
            if other_peer:
                wso = other_peer.ws
                msg: str = "SESSION_END {} {}".format(uid, peer.client_type)
                logger.debug("{} -> {}: {}".format(uid, other_id, msg))
                await wso.send_str(msg)

async def cleanup_room(self, uid: str, room_id: str) -> None:
    """Remove a peer from a room and notify the remaining members.

    Args:
        uid: Peer ID to remove.
        room_id: Room ID to remove from.
    """
    room_peers: Optional[Set[str]] = self.rooms.get(room_id)
    if not room_peers or uid not in room_peers:
        return
    room_peers.remove(uid)
    if not room_peers:
        del self.rooms[room_id]
    # Snapshot: the awaits below can interleave with a concurrent join.
    for pid in list(room_peers):
        peer = self.peers.get(pid)
        if not peer:
            continue
        wsp = peer.ws

        msg = "ROOM_PEER_LEFT {}".format(uid)
        logger.debug("room {}: {} -> {}: {}".format(room_id, uid, pid, msg))
        await wsp.send_str(msg)

def _evict_dead_peer_locked(self, pid: str) -> List[Callable[[], Awaitable[Any]]]:
    """Detach a dead peer from shared state (in-memory only) while holding
    ``self.lock``.

    Mirrors `cleanup_session` and `cleanup_room` without socket I/O. The
    session partner (the server) is told the session ended and keeps its
    socket: it serves every client, and closing it would cascade —
    `remove_peer` on the server closes all clients, including the
    replacement just registered, a self-terminating reconnect.

    Returns:
        Zero-arg coroutine factories for the partner notifications
        (SESSION_END / ROOM_PEER_LEFT / close); the caller MUST run them
        AFTER releasing the lock so a slow partner socket can't stall
        other handshakes.
    """
    deferred: List[Callable[[], Awaitable[Any]]] = []
    peer = self.peers.get(pid)

    if pid in self.sessions:
        other_id = self.sessions[pid]
        del self.sessions[pid]
        if peer is not None:
            other_peer = self.peers.get(other_id)
            if other_peer is not None:
                wso = other_peer.ws
                msg = "SESSION_END {} {}".format(pid, peer.client_type)
                logger.debug("{} -> {}: {}".format(pid, other_id, msg))
                deferred.append(lambda ws=wso, m=msg: ws.send_str(m))

    peer_status = getattr(peer, "peer_status", None) if peer else None
    if peer_status and peer_status != "session":
        room_peers: Optional[Set[str]] = self.rooms.get(peer_status)
        if room_peers and pid in room_peers:
            room_peers.remove(pid)
            if not room_peers:
                del self.rooms[peer_status]
            msg = "ROOM_PEER_LEFT {}".format(pid)
            for other_pid in list(room_peers):
                other_peer = self.peers.get(other_pid)
                if not other_peer:
                    continue
                logger.debug(
                    "room {}: {} -> {}: {}".format(
                        peer_status, pid, other_pid, msg
                    )
                )
                deferred.append(
                    lambda ws=other_peer.ws, m=msg: ws.send_str(m)
                )

    self.peers.pop(pid, None)
    return deferred

async def remove_peer(self, uid: str, ws: Optional[WebSocketResponse] = None) -> None:
    """Remove a peer and clean up its session, room, and socket.

    A server peer leaving closes every client connection, since the server
    owns the media graph they all stream from; a client peer leaving
    closes only its own socket. The server peer leaves when the process
    stops or restarts or the transport switches, none of them a verdict on
    the client, so its clients are closed as going away (1001), which
    they reconnect from, and never with the fatal 4000 the handshake's
    refusals carry. Closes are collected under the lock and awaited after
    it, bounded per socket and best-effort.

    Args:
        uid: Peer ID to remove.
        ws: The socket that ended, when the peer is removed for it: a peer
            that resumed on another socket since stays.
    """
    deferred_closes: List[Callable[[], Awaitable[Any]]] = []
    async with self.lock:
        if ws is not None and getattr(self.peers.get(uid), "ws", ws) is not ws:
            return
        await self.cleanup_session(uid)

        peer = self.peers.get(uid)
        if peer:
            ws = peer.ws
            raddr = peer.raddr
            peer_status = peer.peer_status
            peer_type = peer.peer_type
            client_type = peer.client_type
            if peer_status and peer_status != "session":
                await self.cleanup_room(uid, peer_status)
            if peer_type == "server":
                del self.peers[uid]
                for p in list(self.peers.values()):
                    if p.peer_type == "client":
                        deferred_closes.append(
                            lambda cws=p.ws: cws.close(
                                code=1001,
                                message=b"Server going away.",
                            )
                        )
            else:
                del self.peers[uid]
                deferred_closes.append(
                    lambda cws=ws: cws.close(
                        code=1000, message=b"Connection closed"
                    )
                )
                logger.info(
                    "Disconnected from peer {!r} at {!r} of client_type {!r}".format(
                        uid, raddr, client_type
                    )
                )
    for make_coro in deferred_closes:
        try:
            await asyncio.wait_for(make_coro(), timeout=5)
        except Exception as exc:
            logger.debug("Deferred peer close failed/timed out: {}".format(exc))

async def hold_session(self, uid: str, ws: WebSocketResponse) -> bool:
    """Keep the session of a client whose socket `ws` was cut for
    SESSION_HOLD_S, for a hello naming its resume key, and end it after.

    Returns:
        Whether it is held: a client in no session, one that named no key,
        and one that resumed on another socket since are not.
    """
    async with self.lock:
        peer = self.peers.get(uid)
        if (peer is None or peer.ws is not ws or peer.peer_type != "client"
                or peer.peer_status != "session" or not peer.resume_key):
            return False
        self._holds[uid] = asyncio.create_task(self._end_hold(uid, ws))
    logger.info(f"Signaling socket of peer {uid} cut; its session is held {SESSION_HOLD_S:.0f}s for a resume.")
    return True

async def _end_hold(self, uid: str, ws: WebSocketResponse) -> None:
    """End a held session that no hello resumed in time."""
    await asyncio.sleep(SESSION_HOLD_S)
    self._holds.pop(uid, None)
    await self.remove_peer(uid, ws)
    self._notify_client_presence()

async def close_clients(self) -> None:
    """Close every client socket still registered, as going away (1001),
    for a service that is shutting down: its server peer may already be
    gone, and an open socket keeps the listener's shutdown waiting on it."""
    async with self.lock:
        sockets = [p.ws for p in self.peers.values()
                   if p.peer_type == "client" and not p.ws.closed]
    await asyncio.gather(
        *(asyncio.wait_for(ws.close(code=1001, message=b"Server going away."), timeout=2)
          for ws in sockets),
        return_exceptions=True)

async def peer_connection_handler(
    self,
    ws: WebSocketResponse,
    raddr: str,
    uid: str,
    peer_type: str,
    client_type: Optional[str] = None,
    client_slot: Optional[int] = None,
    client_strict_viewer: Optional[bool] = None,
) -> bool:
    """Serve a registered peer's message loop until it disconnects.

    Relays session messages between paired partners, room messages between
    room members, and processes the SESSION/ROOM commands that establish
    those pairings (message formats in the module docstring). The peer
    was already registered atomically in ``hello_peer``.

    A session relay is accepted in either direction between partners
    (`sessions` is stored client-to-server only) and refused toward an
    unrelated peer or one not in a session. SESSION_START carries the
    caller's secure-mode token (space-free) as a trailing field, omitted
    when absent, so the server side can grant a matching viewer
    read-write collaboration. A room id may not be the "session" status
    sentinel, empty, or contain whitespace.

    Args:
        ws: The peer's WebSocket.
        raddr: Remote address, for logging.
        uid: Peer ID.
        peer_type: Either "server" or "client".
        client_type: For clients, either "viewer" or "controller".
        client_slot: Player slot number (1-4), or -1 when unassigned.
        client_strict_viewer: Whether the client is a strict viewer.

    Returns:
        Whether the socket was cut: it ended with no close frame.
    """
    peer_status = None
    cut = False
    logger.info(
        f"Registered peer {uid} at {raddr} with peer_type {peer_type} and client_type {client_type}"
    )

    try:
        while True:
            msg_obj = await self.recv_msg_ping(ws, raddr)

            if msg_obj.type in [WSMsgType.CLOSE, WSMsgType.CLOSING]:
                logger.info(f"Peer connection closed: {uid!r}")
                break
            elif msg_obj.type in (WSMsgType.CLOSED, WSMsgType.ERROR):
                logger.info(f"Peer connection cut: {uid!r}")
                cut = True
                break

            # autoping is off: answer PING here, feed PONG to the uplink gauge.
            if msg_obj.type == WSMsgType.PING:
                await ws.pong(msg_obj.data)
                continue
            if msg_obj.type == WSMsgType.PONG:
                note_pong(ws, msg_obj.data)
                continue

            if msg_obj.type != WSMsgType.TEXT:
                logger.warning(f"Ignoring non-text message from peer {uid!r}")
                continue
            msg = str(msg_obj.data)

            peer = self.peers.get(uid)
            if not peer:
                logger.error(f"Peer {uid} not found in peers dict")
                break
            peer_status = peer.peer_status
            if peer_status is not None:
                if peer_status == "session":
                    parts = msg.split(maxsplit=1)
                    if len(parts) < 2:
                        logger.warning(f"Malformed session message from {uid}")
                        continue
                    other_id, msg_string = parts
                    other_peer = self.peers.get(other_id)
                    if not other_peer:
                        logger.warning(f"Peer {other_id} not found for session message relay")
                        continue
                    if not (
                        self.sessions.get(uid) == other_id
                        or self.sessions.get(other_id) == uid
                    ):
                        logger.warning(
                            f"Rejecting session relay {uid} -> {other_id}: not session partners"
                        )
                        continue
                    if other_peer.peer_status != "session":
                        logger.warning(
                            f"Rejecting session relay {uid} -> {other_id}: target not in a session"
                        )
                        continue
                    wso = other_peer.ws
                    logger.debug("{} -> {}: {}".format(uid, other_id, msg))
                    msg_string = "{} {}".format(uid, msg_string)
                    try:
                        await wso.send_str(msg_string)
                    except ConnectionError:
                        # A cut partner's lines wait for its resume.
                        if len(other_peer.held) < HELD_LINES_MAX:
                            other_peer.held.append(msg_string)
                elif peer_status:
                    if msg.startswith("ROOM_PEER_MSG"):
                        parts = msg.split(maxsplit=2)
                        if len(parts) < 3:
                            await ws.send_str("ERROR invalid ROOM_PEER_MSG format")
                            continue
                        _, other_id, msg = parts
                        other_peer = self.peers.get(other_id)
                        if not other_peer:
                            await ws.send_str(
                                "ERROR peer {!r} not found".format(other_id)
                            )
                            continue
                        wso = other_peer.ws
                        status = other_peer.peer_status
                        room_id = peer_status
                        if status != room_id:
                            await ws.send_str(
                                "ERROR peer {!r} is not in the room".format(other_id)
                            )
                            continue
                        msg = "ROOM_PEER_MSG {} {}".format(uid, msg)
                        logger.debug(
                            "room {}: {} -> {}: {}".format(room_id, uid, other_id, msg)
                        )
                        await wso.send_str(msg)
                    else:
                        await ws.send_str("ERROR invalid msg, already in room")
                        continue
                else:
                    raise AssertionError("Unknown peer status {!r}".format(peer_status))
            elif msg.startswith("SESSION"):
                logger.debug("{!r} command {!r}".format(uid, msg))
                parts = msg.split(maxsplit=1)
                if len(parts) < 2:
                    logger.warning(f"Malformed session message from {uid}")
                    continue
                _, callee_id = parts
                if callee_id == "server":
                    callee_id = next(
                        (
                            uid
                            for uid, pdata in self.peers.items()
                            if hasattr(pdata, "peer_type")
                            and pdata.peer_type == "server"
                        ),
                        callee_id,
                    )
                # Only the streaming server is called: a session with another
                # page would hand it this page's offer and overwrite its state.
                if getattr(self.peers.get(callee_id), "peer_type", None) != "server":
                    await ws.send_str("ERROR peer server not found")
                    continue
                await ws.send_str("SESSION_OK " + str(callee_id))
                callee_peer = self.peers.get(callee_id)
                if not callee_peer:
                    logger.error(f"Callee peer {callee_id} not found after check")
                    continue
                wsc = callee_peer.ws
                wsc_raadr = callee_peer.raddr
                logger.info(
                    "Session from {!r} ({!r}) to {!r} ({!r})".format(
                        uid, raddr, callee_id, wsc_raadr
                    )
                )
                session_start = "SESSION_START {} {} {} {}".format(
                    uid, client_type, peer.display_id, peer.display_position
                )
                if peer.client_token:
                    session_start += " " + peer.client_token
                if peer.fullcolor_codecs is not None:
                    session_start += " fullcolor=" + ",".join(peer.fullcolor_codecs)
                if peer.tenbit_codecs is not None:
                    session_start += " tenbit=" + ",".join(peer.tenbit_codecs)
                await wsc.send_str(session_start)
                peer.peer_status = peer_status = "session"
                callee_peer.peer_status = "session"
                self.sessions[uid] = callee_id
            elif msg.startswith("ROOM"):
                logger.debug("{!r} command {!r}".format(uid, msg))
                parts = msg.split(maxsplit=1)
                if len(parts) < 2:
                    logger.warning(f"Malformed room message from {uid}")
                    continue
                _, room_id = parts
                if room_id == "session" or room_id.split() != [room_id]:
                    await ws.send_str("ERROR invalid room id {!r}".format(room_id))
                    continue
                if room_id in self.rooms:
                    if uid in self.rooms[room_id]:
                        raise AssertionError(
                            "How did we accept a ROOM command "
                            "despite already being in a room?"
                        )
                else:
                    self.rooms[room_id] = set()
                room_peers = " ".join([pid for pid in self.rooms[room_id]])
                await ws.send_str("ROOM_OK {}".format(room_peers))
                peer.peer_status = peer_status = room_id
                # setdefault: a concurrent cleanup_room may have deleted the room during the await.
                room_set = self.rooms.setdefault(room_id, set())
                room_set.add(uid)
                # Snapshot: peers can join/leave across the awaits.
                for pid in list(room_set):
                    if pid == uid:
                        continue
                    peer = self.peers.get(pid)
                    if not peer:
                        continue
                    wsp = peer.ws
                    msg = "ROOM_PEER_JOINED {}".format(uid)
                    logger.debug("room {}: {} -> {}: {}".format(room_id, uid, pid, msg))
                    await wsp.send_str(msg)
            else:
                logger.info("Ignoring unknown message {!r} from {!r}".format(msg, uid))
    except ConnectionError as e:
        logger.info(f"Peer connection cut: {uid!r} ({e})")
        cut = True
    except Exception as e:
        logger.error(f"Error at connection handler for peer {uid!r}: {e}")
    return cut

def allowed_client_slots(self) -> List[int]:
    """Return the allowed player slot numbers (1 always; 2-4 when enabled)."""
    return [1] + [i for i in range(2, 5) if getattr(self, f"enable_player{i}")]

def _beside(self, peer: Any, client_type: Optional[str], tab_id: Optional[str]) -> bool:
    """Whether a controller joining with `tab_id` stays beside `peer`, a
    controller of the same display, rather than superseding it: with
    sharing on, a page of another tab. A page that names no tab, or the
    peer's own tab reconnecting, supersedes it as before."""
    return (self.enable_sharing and client_type == "controller"
            and getattr(peer, "client_type", None) == "controller"
            and bool(tab_id) and bool(getattr(peer, "tab_id", None)) and peer.tab_id != tab_id)

@staticmethod
def _supersedes(client_type: Optional[str], peer: Any) -> bool:
    """Whether a newcomer of `client_type` may take `peer`'s place: a viewer
    never supersedes a live controller, as on the websockets transport,
    where a viewer's connection takes no display over; it still reaps a
    dead holder and supersedes another viewer."""
    if client_type == "controller" or getattr(peer, "client_type", None) != "controller":
        return True
    peer_ws = getattr(peer, "ws", None)
    return peer_ws is None or peer_ws.closed

def _eviction_storm(self, key: Any) -> bool:
    """Return True when the identity ``key`` (slot/controller) has been
    taken over LIMIT+ times inside the window: two live auto-reconnecting
    pages are trading the session (each takeover closes the other
    mid-handshake). Rejecting the newest claimant breaks the loop and keeps
    the current holder stable; a lone refresh (one takeover per reload)
    never trips this."""
    now = asyncio.get_running_loop().time()
    times = [t for t in self._eviction_times.get(key, ()) if now - t < self._EVICTION_STORM_WINDOW_S]
    if times:
        self._eviction_times[key] = times
    else:
        self._eviction_times.pop(key, None)
    return len(times) >= self._EVICTION_STORM_LIMIT

def _record_eviction(self, key: Any) -> None:
    """Note a takeover of identity `key`, forgetting every identity whose last
    takeover left the window, so the table holds only the identities a storm
    could still trip on, whatever names the hellos carry."""
    now = asyncio.get_running_loop().time()
    recent = {k: [t for t in times if now - t < self._EVICTION_STORM_WINDOW_S]
              for k, times in self._eviction_times.items()}
    self._eviction_times = {k: times for k, times in recent.items() if times}
    self._eviction_times.setdefault(key, []).append(now)

def _secure_token_rejected(self, client_token: Any) -> bool:
    """Secure mode (master token configured) binds streaming access to a
    server-issued token, mirroring the websockets handshake: a client peer must
    present a client_token that maps to a provisioned token. The auth middleware
    skips Basic on WS upgrades in secure mode expecting this gate, so without it
    a client peer would reach signaling unauthenticated."""
    if not app_settings.master_token:
        return False
    if not isinstance(client_token, str):
        return True
    return _lookup_session_token(client_token) is None

def _server_token_rejected(self, server_token: Any) -> bool:
    """The server peer owns the media graph for every client, and a server peer
    leaving closes all of them, so registering as one is an administrative act.
    Only the in-process signaling client may claim it, with this process's
    `SERVER_PEER_SECRET`, and in secure mode anything holding the master token:
    a credential that only lets a page in (Basic, view-only Basic, or none
    behind a proxy) is no proof of being the streaming server."""
    if not server_token or not isinstance(server_token, str):
        return True
    accepted = [SERVER_PEER_SECRET]
    if app_settings.master_token:
        accepted.append(str(app_settings.master_token))
    token = server_token.encode("utf-8")
    return not any(hmac.compare_digest(token, a.encode("utf-8")) for a in accepted)

def _secure_effective_client_type(
    self, client_type: Optional[str], client_token: Any
) -> Optional[str]:
    """Secure mode: the token's provisioned role is authoritative (client_type is
    self-asserted over signaling). Coerce a viewer-role token's 'controller' claim
    to viewer so it can't own a media graph, open a secondary display, or drive
    input — mirroring ws_handler's token-derived role."""
    if not app_settings.master_token or client_type != "controller":
        return client_type
    perms = _lookup_session_token(client_token) if isinstance(client_token, str) else None
    if perms and perms.get("role") == "controller":
        return client_type
    return "viewer"

def _resume_locked(
    self, ws: WebSocketResponse, raddr: str, resume_key: Optional[str], client_type: Optional[str],
    client_token: Optional[str], display_id: str, deferred: List[Callable[[], Awaitable[Any]]],
) -> Optional[Tuple[Tuple[str, str, Optional[str], Optional[int], Optional[bool]], str, List[str]]]:
    """Hand a client's session to the socket of a hello naming its resume
    key, while holding ``self.lock``. The client keeps its peer id and its
    session; a socket of its still open (the cut reached the page alone)
    is closed after the lock is released, without ending the session.

    Returns:
        The hello's result, the session partner's id, and the lines held
        for the client; None where no session of this client is open to
        resume (none of its key, another role, display or token, or the
        partner gone).
    """
    if not resume_key:
        return None
    peer = next((p for p in self.peers.values() if p.peer_type == "client" and p.resume_key
                 and hmac.compare_digest(p.resume_key, resume_key)), None)
    partner = self.peers.get(self.sessions.get(peer.uid, "")) if peer is not None else None
    if (peer is None or peer.peer_status != "session" or getattr(partner, "peer_type", None) != "server"
            or (peer.client_type, peer.display_id, peer.client_token) != (client_type, display_id, client_token)):
        return None
    hold = self._holds.pop(peer.uid, None)
    if hold is not None:
        hold.cancel()
    if not peer.ws.closed:
        deferred.append(lambda old=peer.ws: old.close(code=1000, message=b"Resumed on a new connection."))
    peer.ws, peer.raddr = ws, raddr
    held, peer.held = peer.held, []
    return ((peer.uid, peer.peer_type, peer.client_type, peer.client_slot, peer.client_strict_viewer),
            partner.uid, held)

async def hello_peer(
    self, ws: WebSocketResponse, raddr: str, auth_role_ceiling: Optional[str] = None
) -> Tuple[str, str, Optional[str], Optional[int], Optional[bool]]:
    """Exchange the HELLO handshake, validate, and register the peer.

    Validation, eviction of superseded/dead holders, and registration all
    happen atomically under the lock; the HELLO reply and any deferred
    eviction notifications are sent after release so a slow socket can't
    serialize other handshakes. A HELLO send that fails unregisters the
    peer so a half-open peer never stays in the table.

    Identity collisions resolve newest-wins, as in websockets mode where a
    new display connection supersedes the old one: a page refresh
    reconnects before its old socket is reaped and must not bounce off
    itself, so dead holders are reaped and a live one is closed. The
    identities are the display's sole client in non-sharing mode, a
    player slot in sharing mode (`-1` is the unassigned sentinel, exempt
    from uniqueness) or, in secure mode, the page's token, since its slot
    claim there is only its URL's, and a display's controller — each
    scoped per display so display2 never supersedes the primary — unless
    the takeover-storm breaker (`_eviction_storm`) rejects the claimant. A
    viewer first reaps a dead, unreaped controller and treats it as
    absent rather than pairing with a stale peer; a viewer of the primary
    display needs no controller, since the desktop exists regardless and
    the server starts its capture for a lone viewer (websockets parity),
    while a secondary display exists only through its controller's layout,
    so its viewer is refused without one. Strict shared viewers (the
    `#shared` URL hash) are refused when `enable_shared` is off. A hello
    naming the resume key of a client's open session is that page coming
    back: it takes the session over (`_resume_locked`), is answered
    RESUMED, and meets none of these checks.

    Args:
        ws: The connecting WebSocket.
        raddr: Remote address, for logging.
        auth_role_ceiling: Highest role the transport credential grants;
            "viewer" caps a self-asserted controller.

    Returns:
        Tuple of (peer_id, peer_type, client_type, client_slot,
        client_strict_viewer).

    Raises:
        Exception: Protocol validation failed or the peer was rejected;
            the socket is closed with a reason before raising.
    """
    # With autoping off, a PING/PONG racing the handshake is not an invalid HELLO.
    while True:
        msg_obj = await ws.receive()
        if msg_obj.type == WSMsgType.PING:
            await ws.pong(msg_obj.data)
            continue
        if msg_obj.type == WSMsgType.PONG:
            continue
        break
    if msg_obj.type != WSMsgType.TEXT or not isinstance(msg_obj.data, str):
        await ws.close(code=1002, message=b"invalid protocol")
        raise Exception("Invalid hello message type from {!r}".format(raddr))
    if len(msg_obj.data) > HELLO_MAX_BYTES:
        await ws.close(code=1009, message=b"hello too large")
        raise Exception("Oversized hello ({} bytes) from {!r}".format(len(msg_obj.data), raddr))

    hello = msg_obj.data
    toks = hello.split(maxsplit=2)
    client_type = None
    client_slot = None
    client_strict_viewer = None
    client_tab_id = None
    cc_start_kbps = None
    client_token = None
    server_token = None
    display_id = "primary"
    display_position = "right"
    fullcolor_codecs = None
    tenbit_codecs = None
    page_size = None
    resume_key = None
    resumed = None
    dead_peer_notifications: List[Callable[[], Awaitable[Any]]] = []

    def evict_peer_locked(
        pid: str, peer: Peer, reason: bytes, storm_key: Any = None
    ) -> None:
        """Evict a peer regardless of liveness: a live holder is superseded
        (closed) like websockets mode does for a reconnecting display; a dead
        one is just reaped. Socket I/O is deferred until the lock is released.
        Only live supersedes count toward the takeover-storm window — reaping
        an already-dead holder is routine reconnect housekeeping."""
        peer_ws = getattr(peer, "ws", None)
        if peer_ws is not None and not peer_ws.closed:
            if storm_key is not None:
                self._record_eviction(storm_key)
            dead_peer_notifications.append(
                lambda ws_old=peer_ws, r=reason: ws_old.close(code=4001, message=r)
            )
        dead_peer_notifications.extend(self._evict_dead_peer_locked(pid))

    result = None
    try:
        async with self.lock:
            if len(toks) > 2:
                hello, peer_type, json_metadata_str = toks
                try:
                    json_metadata = json.loads(json_metadata_str)
                    client_type = json_metadata.get("client_type")
                    client_slot = json_metadata.get("client_slot")
                    client_strict_viewer = json_metadata.get("client_strict_viewer")
                    client_token = json_metadata.get("client_token")
                    server_token = json_metadata.get("server_token")
                    named_display = json_metadata.get("display_id")
                    if named_display:
                        if not isinstance(named_display, str) or not DISPLAY_ID_RE.fullmatch(named_display):
                            await ws.close(code=1002, message=b"invalid protocol")
                            raise Exception("Invalid display_id from {!r}".format(raddr))
                        display_id = named_display
                    pos = json_metadata.get("display_position")
                    display_position = pos if pos in ("right", "left", "up", "down") else "right"
                    tab = json_metadata.get("client_tab_id")
                    if isinstance(tab, str) and tab:
                        client_tab_id = tab[:64]
                    cc_start_kbps = json_metadata.get("cc_start_kbps")
                    codecs = json_metadata.get("fullcolor_codecs")
                    if isinstance(codecs, list):
                        fullcolor_codecs = [str(c) for c in codecs if isinstance(c, str) and c.isalnum()]
                    codecs = json_metadata.get("tenbit_codecs")
                    if isinstance(codecs, list):
                        tenbit_codecs = [str(c) for c in codecs if isinstance(c, str) and c.isalnum()]
                    page_size = hello_page_size(json_metadata.get("page_size"))
                    key = json_metadata.get("resume_key")
                    if isinstance(key, str) and 16 <= len(key) <= 64 and key.isascii() and key.isalnum():
                        resume_key = key
                except json.JSONDecodeError as e:
                    await ws.close(code=1002, message=b"invalid protocol")
                    raise Exception("Invalid JSON metadata from {!r}".format(raddr)) from e
                # Coerced to int so collision checks hold (1.0 is slot 1); bool is
                # an int subclass and is rejected rather than read as slot 1.
                if client_slot is not None:
                    if isinstance(client_slot, bool):
                        await ws.close(code=1002, message=b"invalid protocol")
                        raise Exception(
                            "Invalid client_slot {!r} from {!r}".format(
                                client_slot, raddr
                            )
                        )
                    try:
                        client_slot = int(client_slot)
                    except (TypeError, ValueError) as e:
                        await ws.close(code=1002, message=b"invalid protocol")
                        raise Exception(
                            "Invalid client_slot {!r} from {!r}".format(
                                client_slot, raddr
                            )
                        ) from e
            else:
                hello, peer_type = toks

            if hello != "HELLO":
                await ws.close(code=1002, message=b"invalid protocol")
                raise Exception("Invalid hello from {!r}".format(raddr))
            if peer_type not in ("server", "client"):
                await ws.close(code=1002, message=b"invalid protocol")
                raise Exception("Invalid peer type from {!r}".format(raddr))
            if peer_type == "client" and client_type not in ("viewer", "controller"):
                await ws.close(code=1002, message=b"invalid protocol")
                raise Exception("Invalid client type from {!r}".format(raddr))
            if peer_type == "client" and self._secure_token_rejected(client_token):
                await ws.close(code=4001, message=b"Invalid authentication token")
                raise Exception(
                    "Rejecting client from {!r}: missing or invalid token in secure mode".format(raddr)
                )
            if peer_type == "server" and self._server_token_rejected(server_token):
                await ws.close(code=4001, message=b"Invalid authentication token")
                raise Exception(
                    "Rejecting server peer from {!r}: it proved no claim to the server role".format(raddr)
                )

            if peer_type == "client":
                client_type = self._secure_effective_client_type(client_type, client_token)
                if auth_role_ceiling == "viewer" and client_type == "controller":
                    client_type = "viewer"
                resumed = self._resume_locked(ws, raddr, resume_key, client_type, client_token,
                                              display_id, dead_peer_notifications)
            if peer_type == "client" and resumed is None:
                if not self.enable_sharing:
                    existing_clients = [
                        (pid, peer)
                        for pid, peer in self.peers.items()
                        if hasattr(peer, "peer_type") and peer.peer_type == "client"
                        and getattr(peer, "display_id", "primary") == display_id
                        and self._supersedes(client_type, peer)
                    ]
                    if existing_clients and self._eviction_storm(("client", display_id)):
                        await ws.close(
                            code=4000,
                            message=b"Session takeover loop detected; another page holds this session.",
                        )
                        raise Exception(
                            "Rejecting client from {!r}: takeover storm".format(raddr)
                        )
                    for pid, peer in existing_clients:
                        evict_peer_locked(
                            pid, peer, b"Superseded by a new connection.",
                            storm_key=("client", display_id),
                        )
                        logger.info(
                            "Evicting client {!r} for non-sharing reconnect from {!r}".format(
                                pid, raddr
                            )
                        )
                else:
                    allowed_slots = self.allowed_client_slots()
                    if client_slot != -1 and (client_slot not in allowed_slots):
                        await ws.close(
                            code=4000, message=b"Invalid player id provided, check URL."
                        )
                        raise Exception(
                            "Invalid client slot provided {!r}".format(client_slot)
                        )
                    if client_slot != -1:
                        # A page is the only one of its slot, as its hello names it. In
                        # secure mode the token table says who a page is and which slots
                        # it drives, while every page opened without `#playerN` names
                        # slot 1, so a page is the only one of its token instead: the
                        # pages of a collaboration room, each with its own, all stay.
                        by_token = bool(app_settings.master_token)
                        if by_token:
                            key = ("token", display_id, client_token)
                        else:
                            key = ("slot", display_id, client_slot)
                        colliding = [
                            (pid, peer)
                            for pid, peer in self.peers.items()
                            if (getattr(peer, "client_token", None) == client_token if by_token
                                else getattr(peer, "client_slot", None) == client_slot)
                            and getattr(peer, "display_id", "primary") == display_id
                            and not self._beside(peer, client_type, client_tab_id)
                            and self._supersedes(client_type, peer)
                        ]
                        if colliding and self._eviction_storm(key):
                            await ws.close(
                                code=4000,
                                message=(b"Session takeover loop detected; another page holds this session."
                                         if by_token else
                                         b"Player slot takeover loop detected; another page holds this slot."),
                            )
                            # The token is a credential: logs name the page by its address alone.
                            raise Exception(
                                "Rejecting a page of the same token from {!r}: takeover storm".format(raddr)
                                if by_token else
                                "Rejecting slot {!r} claim from {!r}: takeover storm".format(client_slot, raddr)
                            )
                        for pid, peer in colliding:
                            evict_peer_locked(
                                pid, peer,
                                b"Superseded by a new connection." if by_token
                                else b"Superseded by a new connection for this player slot.",
                                storm_key=key,
                            )
                            logger.info(
                                ("Evicting peer {!r} of the same token for reconnect from {!r}".format(pid, raddr)
                                 if by_token else
                                 "Evicting peer {!r} holding slot {!r} for reconnect from {!r}".format(
                                     pid, client_slot, raddr
                                 ))
                                + " (tab {}, the new page's {})".format(
                                    (getattr(peer, "tab_id", None) or "none")[:8], (client_tab_id or "none")[:8])
                            )

                if not self.enable_shared and client_strict_viewer:
                    await ws.close(
                        code=4000, message=b"Strict shared clients are not enabled."
                    )
                    raise Exception(
                        "Strict shared clients are disabled; connection from {!r}".format(
                            raddr
                        )
                    )

                controller_entry = next(
                    (
                        (pid, peer)
                        for pid, peer in self.peers.items()
                        if hasattr(peer, "client_type")
                        and peer.client_type == "controller"
                        and getattr(peer, "display_id", "primary") == display_id
                    ),
                    None,
                )
                peer_controller = controller_entry[1] if controller_entry else None
                if client_type == "controller":
                    # A controller of another tab stays beside the newcomer; the
                    # one it supersedes is its own page's, or any where either
                    # names no tab.
                    rival_entry = next(
                        (
                            (pid, peer)
                            for pid, peer in self.peers.items()
                            if getattr(peer, "client_type", None) == "controller"
                            and getattr(peer, "display_id", "primary") == display_id
                            and not self._beside(peer, client_type, client_tab_id)
                        ),
                        None,
                    )
                    if rival_entry is not None:
                        controller_entry = rival_entry
                        peer_controller = rival_entry[1]
                    elif peer_controller is not None:
                        logger.info("Controller from {!r} joins display {!r} beside {!r} (tab {}, beside tab {})".format(
                            raddr, display_id, controller_entry[0], (client_tab_id or "none")[:8],
                            (getattr(controller_entry[1], "tab_id", None) or "none")[:8]))
                        peer_controller = None
                    if peer_controller is not None:
                        if self._eviction_storm(("controller", display_id)):
                            await ws.close(
                                code=4000,
                                message=b"Session takeover loop detected; another page holds this session.",
                            )
                            raise Exception(
                                "Rejecting controller from {!r}: takeover storm".format(raddr)
                            )
                        assert controller_entry is not None
                        ctrl_pid, ctrl_peer = controller_entry
                        evict_peer_locked(
                            ctrl_pid, ctrl_peer,
                            b"Superseded by a new controller connection.",
                            storm_key=("controller", display_id),
                        )
                        peer_controller = None
                        logger.info(
                            "Evicting controller {!r} for reconnect from {!r} (tab {}, the new page's {})".format(
                                ctrl_pid, raddr, (getattr(ctrl_peer, "tab_id", None) or "none")[:8],
                                (client_tab_id or "none")[:8]
                            )
                        )
                if client_type == "viewer":
                    if peer_controller is not None:
                        assert controller_entry is not None
                        ctrl_pid, ctrl_peer = controller_entry
                        ctrl_ws = getattr(ctrl_peer, "ws", None)
                        if (ctrl_ws is None or ctrl_ws.closed) and ctrl_pid not in self._holds:
                            dead_peer_notifications.extend(
                                self._evict_dead_peer_locked(ctrl_pid)
                            )
                            peer_controller = None
                            logger.info(
                                "Evicting dead controller {!r} ahead of viewer from {!r}".format(
                                    ctrl_pid, raddr
                                )
                            )
                    if not peer_controller and display_id != "primary":
                        await ws.close(
                            code=4000,
                            message=b"No controller detected. A secondary display's viewer requires that display's controller.",
                        )
                        raise Exception(
                            "No controller for display {!r}; rejecting viewer from {!r}".format(
                                display_id, raddr
                            )
                        )

            if resumed is not None:
                result = resumed[0]
            else:
                uid = str(uuid.uuid4())
                puid = "-".join([peer_type, uid])
                self.peers[puid] = Peer(
                    uid=puid,
                    ws=ws,
                    raddr=raddr,
                    peer_type=peer_type,
                    client_type=client_type,
                    client_slot=client_slot,
                    client_strict_viewer=client_strict_viewer,
                    client_token=client_token,
                    display_id=display_id,
                    display_position=display_position,
                    fullcolor_codecs=fullcolor_codecs,
                    tenbit_codecs=tenbit_codecs,
                    tab_id=client_tab_id,
                    cc_start_kbps=cc_start_kbps,
                    page_size=page_size,
                    resume_key=resume_key,
                )
                result = (puid, peer_type, client_type, client_slot, client_strict_viewer)
    finally:
        for make_coro in dead_peer_notifications:
            try:
                await asyncio.wait_for(make_coro(), timeout=5)
            except Exception as exc:
                logger.debug(
                    "Deferred dead-peer notification failed/timed out: {}".format(exc)
                )
    if resumed is not None:
        _, partner_id, held = resumed
        try:
            await ws.send_str("RESUMED " + partner_id)
            for line in held:
                await ws.send_str(line)
        except Exception:
            if not await self.hold_session(result[0], ws):
                await self.remove_peer(result[0], ws)
            raise
        logger.info(f"Peer {result[0]} resumed its session from {raddr}")
        return result
    try:
        await ws.send_str("HELLO")
    except Exception:
        async with self.lock:
            self.peers.pop(result[0], None)
        raise
    return result

async def signaling_handler(
    self, ws: WebSocketResponse, raddr: str, auth_role_ceiling: Optional[str] = None
) -> None:
    """Serve one signaling connection end to end: handshake, message loop,
    and removal on disconnect.

    Args:
        ws: The connecting WebSocket.
        raddr: Remote address, for logging.
        auth_role_ceiling: Highest role the transport-level credential
            grants ("viewer" caps a self-asserted controller in legacy
            basic-auth mode).
    """
    peer_id = None
    try:
        (
            peer_id,
            peer_type,
            client_type,
            client_slot,
            client_strict_viewer,
        ) = await self.hello_peer(ws, raddr, auth_role_ceiling)
    except Exception as e:
        logger.error(f"Error during handshake with peer {raddr}: {e}")
        return
    self._notify_client_presence()

    cut = False
    try:
        cut = await self.peer_connection_handler(
            ws,
            raddr,
            peer_id,
            peer_type,
            client_type,
            client_slot,
            client_strict_viewer,
        )
    except Exception as e:
        logger.error(
            "Error in connection handler for peer {!r}: {}".format(raddr, e),
            exc_info=True,
        )
        await ws.close(code=1002, message=b"internal error")
    finally:
        if peer_id:
            if not (cut and await self.hold_session(peer_id, ws)):
                await self.remove_peer(peer_id, ws)
            self._notify_client_presence()

def _notify_client_presence(self) -> None:
    """Report whether any browser client peer is connected to the presence
    callback; the server's own signaling peer does not count."""
    if self.on_client_presence:
        self.on_client_presence(
            any(p.peer_type == "client" for p in self.peers.values())
        )

async def handle_turn_req(self, request: web.Request) -> web.Response:
    """GET /api/turn: serve the RTC/TURN configuration to clients.

    Who may ask is the auth middleware's call (Basic credentials, or a
    session token in secure mode); this only serves the config. It is the
    exact config the server resolved (`rtc_config`), the single source of
    truth, with no per-request credential path: the client always
    negotiates with the same ICE servers the server uses. When the HMAC
    method is active, `generate_rtc_config()` mints the username with a
    generic default so it is never a bare `<expiry>:`.

    Returns:
        The RTC configuration as JSON, or 404 when none is resolved.
    """
    path = request.path

    if self.rtc_config:
        data = self.rtc_config
        if isinstance(data, str):
            data = data.encode("utf-8")
        return web.Response(status=200, body=data, content_type="application/json")

    logger.warning("HTTP GET {} 404 NOT FOUND - Missing RTC config".format(path))
    return web.Response(status=404, text="404 NOT FOUND")