Files
hockey_new/hockey_data/match_access.py

495 lines
28 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Match authority, fencing and atomic optimistic writes.
Read channels are not credentials for controlling a match. A lease belongs to
one authenticated user AND one browser participant. No read/join auto-claims it.
"""
from __future__ import annotations
import hashlib
import inspect
import json
import re
from datetime import datetime, timedelta
from functools import wraps
from typing import Any, Callable
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy import delete, select, update
from .auth_bridge import HockeyUser
from .async_work import db_endpoint, run_db
from .account_activity import account_is_idle, lock_account
from .models import Game, MatchCommandReceipt, MatchControlLease, MatchParticipant, OperatorSession, VmixDevice
LEASE_SECONDS = 90
PARTICIPANT_SECONDS = 120
_IDENTIFIER = re.compile(r"[A-Za-z0-9_-]{16,64}\Z")
def fail(status: int, code: str, message: str, **extra: Any) -> None:
raise HTTPException(status_code=status, detail={"code": code, "message": message, **extra})
def participant_id(request: Request, *, required: bool = True) -> str:
value = request.headers.get("X-Hockey-Participant", "")
if not _IDENTIFIER.fullmatch(value):
if required:
fail(428, "participant_required", "Обновите страницу: не указан участник матча")
return ""
return value
def header_int(request: Request, name: str) -> int:
value = request.headers.get(name, "")
if not value.isascii() or not value.isdigit() or len(value) > 16:
fail(428, "version_required", "Обновите состояние матча перед изменением")
return int(value)
def _iso(value: datetime | None) -> str:
return value.isoformat() + "Z" if value else ""
class MatchAccess:
def __init__(self, database: Any) -> None:
self.database = database
@staticmethod
def lock_account(session: Any, user_id: str) -> None:
"""Serialize primary-channel switches with claims in the same account."""
lock_account(session, user_id)
if account_is_idle(session, user_id, datetime.utcnow()):
fail(401, "account_inactive", "Два часа без действий. Войдите снова и выберите матч")
@staticmethod
def _eligible(user: HockeyUser) -> bool:
return bool(user.is_active and (user.is_admin or user.role == "operator"))
@staticmethod
def _active(lease: MatchControlLease | None, now: datetime) -> bool:
return bool(lease and lease.owner_participant_id and lease.expires_at and lease.expires_at > now)
@staticmethod
def _check_game(session: Any, game_id: str) -> None:
if session.scalar(select(Game.external_id).where(Game.external_id == game_id)) is None:
fail(404, "game_not_found", "Матч не найден")
def _lock(self, session: Any, game_id: str, *, create: bool = False) -> MatchControlLease | None:
if create:
self._check_game(session, game_id)
# Race-safe initialization, including when two browsers claim first.
if session.get_bind().dialect.name == "postgresql":
from sqlalchemy.dialects.postgresql import insert
elif session.get_bind().dialect.name == "sqlite":
from sqlalchemy.dialects.sqlite import insert
else:
raise RuntimeError("Unsupported match authority database")
session.execute(insert(MatchControlLease).values(game_external_id=game_id).on_conflict_do_nothing())
return session.scalar(select(MatchControlLease).where(
MatchControlLease.game_external_id == game_id
).with_for_update().execution_options(populate_existing=True))
def _participant(self, session: Any, game_id: str, pid: str, user: HockeyUser) -> MatchParticipant:
row = session.get(MatchParticipant, (game_id, pid))
if row is None or row.user_id != user.id:
fail(403, "participant_not_joined", "Сначала откройте этот матч в своём браузере")
return row
def _owned(self, lease: MatchControlLease | None, user: HockeyUser, pid: str, generation: int) -> MatchControlLease:
if not self._eligible(user) or not self._active(lease, datetime.utcnow()):
fail(403, "view_only", "Режим зрителя: управление матчем недоступно")
if lease.owner_user_id != user.id or lease.owner_participant_id != pid:
fail(403, "view_only", "Матчем управляет другой участник")
if lease.generation != generation:
fail(409, "control_changed", "Управление изменилось. Обновите состояние матча")
return lease
def _payload(self, session: Any, game_id: str, user: HockeyUser, pid: str) -> dict[str, Any]:
now = datetime.utcnow()
lease = session.get(MatchControlLease, game_id)
active = self._active(lease, now)
own = bool(active and lease.owner_user_id == user.id and lease.owner_participant_id == pid and self._eligible(user))
participants = session.scalars(select(MatchParticipant).where(
MatchParticipant.game_external_id == game_id,
MatchParticipant.last_seen_at >= now - timedelta(seconds=PARTICIPANT_SECONDS),
).order_by(MatchParticipant.last_seen_at.desc()).limit(64))
return {
"game_id": game_id, "participant_id": pid,
"role": "controller" if own else "viewer", "can_control": own,
"control_device_id": lease.owner_device_id if own else "",
"can_claim": self._eligible(user) and not active,
"can_resume": bool(active and lease.owner_user_id == user.id and self._eligible(user) and not own),
"can_force_claim": bool(user.is_admin and not own),
"generation": lease.generation if lease else 0,
"revision": lease.revision if lease else 0,
"owner": {"name": lease.owner_name, "participant_id": lease.owner_participant_id,
"is_self": lease.owner_user_id == user.id} if active else None,
"expires_at": _iso(lease.expires_at) if active else "",
"lease_remaining_ms": max(0, int((lease.expires_at - now).total_seconds() * 1000)) if active else 0,
"server_time": _iso(now),
"participants": [{"participant_id": p.participant_id, "name": p.display_name,
"can_claim": p.can_claim} for p in participants],
}
def snapshot(self, game_id: str, user: HockeyUser, pid: str = "") -> dict[str, Any]:
with self.database.session() as session:
self._check_game(session, game_id)
return self._payload(session, game_id, user, pid)
def join(self, game_id: str, user: HockeyUser, pid: str, device_id: str = "") -> dict[str, Any]:
with self.database.session() as session:
self._check_game(session, game_id)
now = datetime.utcnow()
# Participant metadata is deliberately separate from game events.
row = session.get(MatchParticipant, (game_id, pid))
if row is not None and row.user_id != user.id:
fail(403, "participant_mismatch", "Участник принадлежит другому пользователю")
if device_id:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or device.wfl_user_id != user.id:
fail(403, "device_mismatch", "Agent не принадлежит вашему аккаунту")
if row is None:
row = MatchParticipant(game_external_id=game_id, participant_id=pid, user_id=user.id)
session.add(row)
row.display_name = (user.display_name or user.login)[:255]
row.can_claim = self._eligible(user)
row.device_id = device_id
row.last_seen_at = now
session.execute(delete(MatchParticipant).where(
MatchParticipant.game_external_id == game_id,
MatchParticipant.last_seen_at < now - timedelta(days=1),
))
session.flush()
return self._payload(session, game_id, user, pid)
def change(self, game_id: str, user: HockeyUser, request: Request, action: str,
*, target_id: str = "", force: bool = False) -> dict[str, Any]:
pid = participant_id(request)
generation = header_int(request, "X-Hockey-Control-Generation")
with self.database.atomic_session() as session:
self.lock_account(session, user.id)
member = self._participant(session, game_id, pid, user)
lease = self._lock(session, game_id, create=True)
now = datetime.utcnow()
old_generation = lease.generation
if generation != old_generation:
fail(409, "control_changed", "Обновите список участников перед передачей управления")
if action == "claim":
if not self._eligible(user):
fail(403, "view_only", "Ваш аккаунт имеет доступ только к просмотру")
if self._active(lease, now) and not (force and user.is_admin):
fail(409, "control_busy", "Матч уже находится под управлением оператора")
target = member
elif action == "resume":
# An explicit move between this account's browsers. Matching
# display names never grant authority; the authenticated ID does.
if not self._eligible(user) or not self._active(lease, now) or lease.owner_user_id != user.id:
fail(403, "view_only", "Продолжить здесь можно только управление своего аккаунта")
target = member
if not target.device_id and lease.owner_device_id:
device = session.scalar(select(VmixDevice).where(
VmixDevice.device_uuid == lease.owner_device_id,
VmixDevice.wfl_user_id == user.id,
))
if device is not None:
target.device_id = device.device_uuid
else:
self._owned(lease, user, pid, generation)
target = member
if action == "transfer":
target = session.get(MatchParticipant, (game_id, target_id))
if not target or not target.can_claim or target.last_seen_at < now - timedelta(seconds=PARTICIPANT_SECONDS):
fail(409, "target_unavailable", "Выбранный участник отключён или не может управлять")
values: dict[str, Any] = {"updated_at": now}
if action in {"claim", "transfer", "resume"} and target.device_id:
# Serialize claims that target the same physical Agent, even
# across different matches of one account.
session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == target.device_id).with_for_update())
busy = session.scalar(select(MatchControlLease.game_external_id).where(
MatchControlLease.game_external_id != game_id,
MatchControlLease.owner_device_id == target.device_id,
MatchControlLease.expires_at > now,
).limit(1))
if busy:
fail(409, "device_in_control", "Agent уже используется для управления другим матчем")
if action == "heartbeat":
values["expires_at"] = now + timedelta(seconds=LEASE_SECONDS)
else:
values.update(generation=old_generation + 1, revision=lease.revision + 1)
if action == "release":
values.update(owner_participant_id="", owner_user_id="", owner_name="", owner_device_id="", expires_at=None)
else:
values.update(owner_participant_id=target.participant_id, owner_user_id=target.user_id,
owner_name=target.display_name, owner_device_id=target.device_id,
expires_at=now + timedelta(seconds=LEASE_SECONDS))
changed = session.execute(update(MatchControlLease).where(
MatchControlLease.game_external_id == game_id,
MatchControlLease.generation == old_generation,
MatchControlLease.revision == lease.revision,
).values(**values).execution_options(synchronize_session=False))
if changed.rowcount != 1:
fail(409, "control_changed", "Управление уже изменилось в другом браузере")
member.last_seen_at = now
session.flush()
session.expire(lease)
return self._payload(session, game_id, user, pid)
def leave(self, game_id: str, user: HockeyUser, request: Request) -> dict[str, Any]:
pid = participant_id(request)
with self.database.session() as session:
lease = session.get(MatchControlLease, game_id)
if (self._active(lease, datetime.utcnow()) and lease.owner_user_id == user.id
and lease.owner_participant_id == pid):
fail(409, "release_required", "Сначала завершите управление матчем")
session.execute(delete(MatchParticipant).where(
MatchParticipant.game_external_id == game_id,
MatchParticipant.participant_id == pid, MatchParticipant.user_id == user.id,
))
return {"ok": True}
def require_controller(self, request: Request, user: HockeyUser, game_id: str = "") -> str:
game_id = str(game_id or request.headers.get("X-Hockey-Game", "")).strip()
if not game_id:
fail(428, "game_required", "Сначала выберите матч")
supplied = request.headers.get("X-Hockey-Game", game_id)
if supplied != game_id:
fail(409, "game_changed", "Команда относится к другому матчу")
with self.database.session() as session:
self._owned(session.get(MatchControlLease, game_id), user, participant_id(request),
header_int(request, "X-Hockey-Control-Generation"))
return game_id
def game_for_session(self, token: str, user: HockeyUser) -> str:
if not token:
fail(428, "session_required", "Откройте матч перед выполнением команды")
with self.database.session() as session:
row = session.scalar(select(OperatorSession).where(
OperatorSession.session_token == token, OperatorSession.wfl_user_id == user.id,
OperatorSession.status == "active"))
if row is None or not row.game_external_id:
fail(404, "session_not_found", "Матчевая сессия не найдена")
return row.game_external_id
def require_device_controller(self, request: Request, user: HockeyUser, device_id: str) -> str:
with self.database.session() as session:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
if device is None or device.wfl_user_id != user.id:
fail(403, "device_mismatch", "Agent не принадлежит вашему аккаунту")
game_id = str(device.current_match_external_id or "")
return self.require_controller(request, user, game_id)
def session_device(self, token: str, user: HockeyUser) -> str:
with self.database.session() as session:
row = session.scalar(select(OperatorSession).where(
OperatorSession.session_token == token, OperatorSession.wfl_user_id == user.id,
OperatorSession.status == "active"))
return str(row.vmix_device_uuid or "") if row else ""
def checked_context(self, request: Request, user: HockeyUser, supplied: dict[str, Any]) -> dict[str, Any]:
context = dict(supplied or {})
game_id = self.require_controller(request, user, str(context.get("game_id") or ""))
if context.get("session_token"):
if self.game_for_session(str(context["session_token"]), user) != game_id:
fail(409, "game_changed", "Контекст относится к другой сессии")
self.protect_session(request, user, str(context["session_token"]))
if not context.get("device_id"):
context["device_id"] = self.session_device(str(context["session_token"]), user)
if context.get("device_id"):
if self.require_device_controller(request, user, str(context["device_id"])) != game_id:
fail(409, "game_changed", "Agent назначен другому матчу")
context.update(game_id=game_id, account_id=user.id)
return context
def protect_device(self, request: Request, user: HockeyUser, device_id: str) -> None:
"""A second tab of the same account cannot reassign the on-air Agent."""
if not device_id:
return
pid = participant_id(request)
with self.database.session() as session:
lease = session.scalar(select(MatchControlLease).where(
MatchControlLease.owner_user_id == user.id,
MatchControlLease.owner_device_id == device_id,
MatchControlLease.owner_participant_id != pid,
MatchControlLease.expires_at > datetime.utcnow(),
))
if lease:
fail(403, "device_in_control", "Этот Agent используется управляющим оператором в другом браузере")
def channel_participant(self, request: Request, user: HockeyUser) -> str:
pid = participant_id(request)
with self.database.session() as session:
other = session.scalar(select(MatchControlLease.game_external_id).where(
MatchControlLease.owner_user_id == user.id,
MatchControlLease.owner_participant_id != pid,
MatchControlLease.expires_at > datetime.utcnow(),
).limit(1))
# Reuse a secondary channel on reload, even after the owner leaves.
existing = session.scalar(select(OperatorSession.id).where(
OperatorSession.wfl_user_id == user.id,
OperatorSession.browser_participant_id == pid,
).limit(1))
return pid if other or existing else ""
def bind_controller_device(self, request: Request, user: HockeyUser, game_id: str, device_id: str) -> None:
pid = participant_id(request)
with self.database.atomic_session() as session:
lease = self._lock(session, game_id)
if not (self._active(lease, datetime.utcnow()) and lease.owner_user_id == user.id
and lease.owner_participant_id == pid):
return
if device_id:
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id).with_for_update())
if not device or device.wfl_user_id != user.id:
fail(403, "device_mismatch", "Agent не принадлежит вашему аккаунту")
busy = session.scalar(select(MatchControlLease.game_external_id).where(
MatchControlLease.game_external_id != game_id,
MatchControlLease.owner_device_id == device_id,
MatchControlLease.expires_at > datetime.utcnow(),
).limit(1))
if busy:
fail(409, "device_in_control", "Agent используется в другом матче")
lease.owner_device_id = device_id
member = self._participant(session, game_id, pid, user)
member.device_id = device_id
def protect_session(self, request: Request, user: HockeyUser, token: str) -> None:
pid = participant_id(request)
with self.database.session() as session:
row = session.scalar(select(OperatorSession).where(
OperatorSession.session_token == token, OperatorSession.wfl_user_id == user.id))
if row is None:
fail(404, "session_not_found", "Матчевая сессия не найдена")
if row.browser_participant_id and row.browser_participant_id != pid:
fail(403, "session_in_use", "Канал принадлежит другому браузеру")
lease = session.get(MatchControlLease, row.game_external_id)
if (self._active(lease, datetime.utcnow()) and lease.owner_user_id == user.id
and lease.owner_participant_id != pid and not row.browser_participant_id):
fail(403, "session_in_use", "Канал используется управляющим оператором")
async def execute(self, request: Request, user: HockeyUser, game_id: str,
operation: Callable[[], Any]) -> dict[str, Any]:
pid = participant_id(request)
generation = header_int(request, "X-Hockey-Control-Generation")
expected = header_int(request, "X-Hockey-Revision")
command_id = request.headers.get("X-Hockey-Command-Id", "")
if not _IDENTIFIER.fullmatch(command_id):
fail(428, "command_required", "Не указан идентификатор изменения")
digest = hashlib.sha256(request.method.encode() + str(request.url.path).encode()
+ request.url.query.encode() + await request.body()).hexdigest()
return await run_db(self._execute_sync, request, user, game_id, operation,
pid, generation, expected, command_id, digest)
def _execute_sync(self, request: Request, user: HockeyUser, game_id: str,
operation: Callable[[], Any], pid: str, generation: int,
expected: int, command_id: str, digest: str) -> dict[str, Any]:
with self.database.atomic_session() as session:
lease = self._owned(self._lock(session, game_id), user, pid, generation)
receipt = session.get(MatchCommandReceipt, (game_id, command_id))
if receipt:
if receipt.request_hash != digest or receipt.generation != generation:
fail(409, "command_reused", "Идентификатор команды уже использован")
if receipt.revision != lease.revision:
fail(409, "already_applied", "Команда уже выполнена; получите актуальное состояние", revision=lease.revision)
result = json.loads(receipt.response_json)
result["control_access"] = self._payload(session, game_id, user, pid)
return result
if lease.revision != expected:
fail(409, "stale_revision", "Состояние матча изменилось. Данные обновлены; повторите действие", revision=lease.revision)
changed = session.execute(update(MatchControlLease).where(
MatchControlLease.game_external_id == game_id,
MatchControlLease.revision == expected,
MatchControlLease.generation == generation,
MatchControlLease.owner_user_id == user.id,
MatchControlLease.owner_participant_id == pid,
MatchControlLease.expires_at > datetime.utcnow(),
).values(revision=expected + 1, updated_at=datetime.utcnow()).execution_options(synchronize_session=False))
if changed.rowcount != 1:
fail(409, "stale_revision", "Состояние матча уже изменено")
session.expire(lease)
result = operation()
result = dict(result)
session.flush()
result["revision"] = expected + 1
result["control_access"] = self._payload(session, game_id, user, pid)
encoded = json.dumps(result, ensure_ascii=False, separators=(",", ":"))
if len(encoded) > 2_000_000:
fail(413, "result_too_large", "Слишком большой результат изменения")
session.add(MatchCommandReceipt(game_external_id=game_id, command_id=command_id,
generation=generation, revision=expected + 1, request_hash=digest, response_json=encoded))
session.execute(delete(MatchCommandReceipt).where(
MatchCommandReceipt.game_external_id == game_id,
MatchCommandReceipt.revision <= expected - 64,
))
return result
def command_route(self, fn: Callable[..., Any]) -> Callable[..., Any]:
signature = inspect.signature(fn, eval_str=True)
@wraps(fn)
async def guarded(*args: Any, **kwargs: Any) -> Any:
values = signature.bind(*args, **kwargs).arguments
return await self.execute(values["request"], values["user"], values["external_id"],
lambda: fn(*args, **kwargs))
guarded.__signature__ = signature
return guarded
class JoinPayload(BaseModel):
device_id: str = Field(default="", max_length=128)
class ClaimPayload(BaseModel):
force: bool = False
class TransferPayload(BaseModel):
participant_id: str = Field(min_length=16, max_length=64)
def create_match_access_router(access: MatchAccess, auth_dependency: Callable[..., Any]) -> APIRouter:
router = APIRouter()
@router.get("/games/{external_id}/access")
@db_endpoint
def status(external_id: str, request: Request, user: HockeyUser = Depends(auth_dependency)):
return access.snapshot(external_id, user, participant_id(request))
@router.post("/games/{external_id}/access/join")
@db_endpoint
def join(external_id: str, request: Request, payload: JoinPayload, user: HockeyUser = Depends(auth_dependency)):
return access.join(external_id, user, participant_id(request), payload.device_id)
@router.post("/games/{external_id}/access/claim")
@db_endpoint
def claim(external_id: str, request: Request, payload: ClaimPayload, user: HockeyUser = Depends(auth_dependency)):
return access.change(external_id, user, request, "claim", force=payload.force)
@router.post("/games/{external_id}/access/resume")
@db_endpoint
def resume(external_id: str, request: Request, user: HockeyUser = Depends(auth_dependency)):
return access.change(external_id, user, request, "resume")
@router.post("/games/{external_id}/access/leave")
@db_endpoint
def leave(external_id: str, request: Request, user: HockeyUser = Depends(auth_dependency)):
return access.leave(external_id, user, request)
@router.post("/games/{external_id}/access/heartbeat")
@db_endpoint
def heartbeat(external_id: str, request: Request, user: HockeyUser = Depends(auth_dependency)):
return access.change(external_id, user, request, "heartbeat")
@router.post("/games/{external_id}/access/release")
@db_endpoint
def release(external_id: str, request: Request, user: HockeyUser = Depends(auth_dependency)):
return access.change(external_id, user, request, "release")
@router.post("/games/{external_id}/access/transfer")
@db_endpoint
def transfer(external_id: str, request: Request, payload: TransferPayload, user: HockeyUser = Depends(auth_dependency)):
return access.change(external_id, user, request, "transfer", target_id=payload.participant_id)
return router