From 36a9862a946c2a15a34de59d2ed8b4b19c221f6f Mon Sep 17 00:00:00 2001 From: gglamer Date: Sun, 27 Sep 2026 07:56:12 +0000 Subject: [PATCH] =?UTF-8?q?refactor:=20=D0=BE=D0=B1=D1=89=D0=B8=D0=B9=20?= =?UTF-8?q?=D0=B2=D1=85=D0=BE=D0=B4=20=D0=B2=20=D1=81=D0=BE=D0=BA=D0=B5?= =?UTF-8?q?=D1=82=20=D0=B7=D0=B0=D0=BD=D1=8F=D1=82=D0=B8=D1=8F=20=D0=B8=20?= =?UTF-8?q?=D0=B2=D1=8B=D0=BF=D0=BE=D0=BB=D0=BD=D0=B5=D0=BD=D0=B8=D0=B5=20?= =?UTF-8?q?=D0=BA=D0=BE=D0=BC=D0=B0=D0=BD=D0=B4=D1=8B=20=D1=81=20fencing?= =?UTF-8?q?=20=D0=B2=D0=BC=D0=B5=D1=81=D1=82=D0=BE=20=D0=BA=D0=BE=D0=BF?= =?UTF-8?q?=D0=B8=D0=B9=20=D0=B2=20=D1=87=D0=B5=D1=82=D1=8B=D1=80=D1=91?= =?UTF-8?q?=D1=85=20=D0=BA=D0=B0=D0=BD=D0=B0=D0=BB=D0=B0=D1=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/http/sessions.py | 50 ++-- backend/app/api/ws/call.py | 76 ++---- backend/app/api/ws/control.py | 59 +--- backend/app/api/ws/observe.py | 60 +---- backend/app/api/ws/session.py | 160 +++++++++++ backend/app/api/ws/station.py | 95 +------ backend/app/session/access.py | 30 +++ backend/tests/test_db.py | 8 +- backend/tests/test_route_auth_contract.py | 12 +- backend/tests/test_session_access.py | 5 +- backend/tests/test_session_socket.py | 312 ++++++++++++++++++++++ backend/tests/test_ws_ownership.py | 17 +- 12 files changed, 602 insertions(+), 282 deletions(-) create mode 100644 backend/app/api/ws/session.py create mode 100644 backend/app/session/access.py create mode 100644 backend/tests/test_session_socket.py diff --git a/backend/app/api/http/sessions.py b/backend/app/api/http/sessions.py index 02ff444..9f3defd 100644 --- a/backend/app/api/http/sessions.py +++ b/backend/app/api/http/sessions.py @@ -27,6 +27,7 @@ from app.domain.statuses import SERVICE_STATUS_LABELS, StationSnapshot, current from app.domain.timers import TimerCode from app.scoring.export import to_csv, to_pdf from app.scoring.report import build as build_report +from app.session.access import can_access from app.session.checkpoint import load_state from app.session.finish import override_score, scoring_scenario from app.session.hub import hub @@ -328,16 +329,23 @@ async def create(body: SessionCreate, request: Request, db: AsyncSession = Depen return _out(session) +def _require_access(who, lesson, what: str = "session") -> None: + """Отказ по правилу `can_access`: курсанту — 403, чужому преподавателю + занятие «не существует» — 404, владелец не раскрывается.""" + if can_access(who, lesson): + return + if who.role is Role.TRAINEE: + raise HTTPException(status_code=403, detail=f"not_your_{what}") + raise HTTPException(status_code=404, detail=f"{what}_not_found") + + @router.get("/{session_id}", response_model=SessionOut) async def read(session_id: UUID, request: Request, db: AsyncSession = Depends(get_session)) -> SessionOut: who = require(request) session = await repo.get_session(db, session_id) if session is None: raise HTTPException(status_code=404, detail="session_not_found") - if who.role is Role.TRAINEE and session.trainee_id != who.trainee_id: - raise HTTPException(status_code=403, detail="not_your_session") - if who.role is Role.INSTRUCTOR and session.owner_login != who.login: - raise HTTPException(status_code=404, detail="session_not_found") + _require_access(who, session) return _out(session) @@ -358,10 +366,7 @@ async def checklist(session_id: UUID, request: Request) -> list[ChecklistItemOut state = hub.get(session_id) if state is None: raise HTTPException(status_code=404, detail="session_not_found") - if who.role is Role.TRAINEE and state.trainee_id != who.trainee_id: - raise HTTPException(status_code=403, detail="not_your_session") - if who.role is Role.INSTRUCTOR and state.owner_login != who.login: - raise HTTPException(status_code=404, detail="session_not_found") + _require_access(who, state) if not state.ended: raise HTTPException(status_code=409, detail="call_not_ended") scenario = scoring_scenario(state) @@ -417,10 +422,7 @@ async def _report_data( state = None scenario = None if state is not None and scenario is not None: - if who.role is Role.INSTRUCTOR and state.owner_login != who.login: - raise HTTPException(status_code=404, detail="session_not_found") - if who.role is Role.TRAINEE and state.trainee_id != who.trainee_id: - raise HTTPException(status_code=403, detail="not_your_session") + _require_access(who, state) if who.role is Role.TRAINEE and state.exercise is Exercise.CALL and not state.self_assessed: raise HTTPException(status_code=409, detail="self_assessment_required") if state.score is None: @@ -443,10 +445,7 @@ async def _report_data( session = await repo.get_session(db, session_id) if session is None: raise HTTPException(status_code=404, detail="session_not_found") - if who.role is Role.TRAINEE and session.trainee_id != who.trainee_id: - raise HTTPException(status_code=403, detail="not_your_session") - if who.role is Role.INSTRUCTOR and session.owner_login != who.login: - raise HTTPException(status_code=404, detail="session_not_found") + _require_access(who, session) score = await db.scalar(select(Score).where(Score.session_id == session_id)) if score is None: raise HTTPException(status_code=409, detail="score_not_ready") @@ -516,20 +515,13 @@ async def recording(session_id: UUID, request: Request, db: AsyncSession = Depen state = hub.get(session_id) if state is None: raise HTTPException(status_code=404, detail="session_not_found") - owner_id = state.trainee_id - owner_login = state.owner_login - ended = state.ended + lesson, ended = state, state.ended else: row = await repo.get_session(db, session_id) if row is None: raise HTTPException(status_code=404, detail="session_not_found") - owner_id = row.trainee_id - owner_login = row.owner_login - ended = row.ended_at is not None - if who.role is Role.TRAINEE and (owner_id is None or owner_id != who.trainee_id): - raise HTTPException(status_code=403, detail="not_your_recording") - if who.role is Role.INSTRUCTOR and owner_login != who.login: - raise HTTPException(status_code=404, detail="recording_not_found") + lesson, ended = row, row.ended_at is not None + _require_access(who, lesson, "recording") if not ended: raise HTTPException(status_code=409, detail="recording_not_ready") path = recording_path(session_id) @@ -562,8 +554,7 @@ async def override( ) state = hub.get(session_id) if state is not None: - if state.owner_login != who.login: - raise HTTPException(status_code=404, detail="session_not_found") + _require_access(who, state) if state.score is None: raise HTTPException(status_code=409, detail="score_not_ready") scenario = scoring_scenario(state) @@ -577,8 +568,9 @@ async def override( if db is None: raise HTTPException(status_code=404, detail="session_not_found") session = await repo.get_session(db, session_id) - if session is None or session.owner_login != who.login: + if session is None: raise HTTPException(status_code=404, detail="session_not_found") + _require_access(who, session) score = await db.scalar(select(Score).where(Score.session_id == session_id)) if score is None: raise HTTPException(status_code=409, detail="score_not_ready") diff --git a/backend/app/api/ws/call.py b/backend/app/api/ws/call.py index 90d2956..2d0f0d4 100644 --- a/backend/app/api/ws/call.py +++ b/backend/app/api/ws/call.py @@ -14,7 +14,7 @@ from uuid import UUID, uuid4 from fastapi import APIRouter, WebSocket, WebSocketDisconnect from pydantic import TypeAdapter, ValidationError -from app.api.auth import principal_of, websocket_origin_allowed +from app.api.ws.session import pump, run_command, session_socket from app.domain.events import ( BgStart, CallEnded, @@ -43,7 +43,7 @@ from app.domain.roles import Role from app.scenarios import store from app.session.dds import prepare_handoff_queue from app.session.finish import end_session, refresh_archived_report, release_score -from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.hub import hub from app.session.state import now_utc from app.session.store import ( HintRecorded, @@ -381,66 +381,16 @@ def _start_voice(session_id: UUID, state, *, initial_statement: bool = True) -> state.voice.speak(scenario.first_line, state.persona.mood) -async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None: - while True: - item = await queue.get() - # Бинарь — звук звонящего, без обёртки JSON (docs/arch/CONTRACT.md). - if isinstance(item, bytes): - await ws.send_bytes(item) - else: - await ws.send_text(item.model_dump_json()) - if (isinstance(item, ErrorEvent) and item.code is ErrorKind.INTERNAL - and item.message == LEASE_FENCED_MESSAGE): - await ws.close(code=1012) - return - - -async def _reject(ws: WebSocket, message: str) -> None: - """Отказ до входа в цикл: сокет закрывается с объяснением, а не молча.""" - await ws.send_text( - ErrorEvent(code=ErrorKind.FORBIDDEN, message=message).model_dump_json() - ) - await ws.close() - - @router.websocket("/ws/call/{session_id}") async def call(ws: WebSocket, session_id: UUID) -> None: - if not websocket_origin_allowed(ws): - await ws.close(code=1008) - return - if hub.is_lease_fenced(session_id): - await ws.close(code=1012) - return - await ws.accept() - # АРМ курсанта. Преподаватель допущен, чтобы показать приём вызова группе. - who = principal_of(ws) - if who is None or who.role not in (Role.TRAINEE, Role.INSTRUCTOR): - await _reject(ws, "Недостаточно прав для этого экрана") - return - - state = hub.get(session_id) - if state is None: - await ws.send_text( - ErrorEvent( - code=ErrorKind.SESSION_NOT_FOUND, message="Занятие ещё не запущено преподавателем" - ).model_dump_json() - ) - await ws.close() - return - if who.role is Role.INSTRUCTOR and state.owner_login != who.login: - await ws.send_text( - ErrorEvent( - code=ErrorKind.SESSION_NOT_FOUND, message="Занятие ещё не запущено преподавателем" - ).model_dump_json() - ) - await ws.close() - return - if who.role is Role.TRAINEE and ( - state.trainee_id is None or state.trainee_id != who.trainee_id - ): - await _reject(ws, "Занятие не назначено этому обучающемуся") + entered = await session_socket( + ws, session_id, (Role.TRAINEE, Role.INSTRUCTOR), + not_found="Занятие ещё не запущено преподавателем", + ) + if entered is None: return + _who, state = entered with hub.trainee(session_id) as queue: if state.exercise is Exercise.CARD: @@ -476,7 +426,7 @@ async def call(ws: WebSocket, session_id: UUID) -> None: if state.recorder is None: state.recorder = start_recording(session_id) _start_voice(session_id, state, initial_statement=False) - writer = asyncio.create_task(_pump(ws, queue)) + writer = asyncio.create_task(pump(ws, queue)) try: while True: message = await ws.receive() @@ -509,8 +459,12 @@ async def call(ws: WebSocket, session_id: UUID) -> None: ErrorEvent(code=ErrorKind.UNSUPPORTED_EVENT, message=str(payload)[:200]), ) continue - async with hub.operation(session_id): - await _handle(session_id, state, event) + # `_command_id` у `kio.patch` подтверждается эхом правки, а не + # `CommandAck`: его нет в событиях курсанта, повторы не отсеиваются. + if not await run_command( + ws, session_id, lambda: _handle(session_id, state, event), sender=writer, + ): + return except WebSocketDisconnect: return finally: diff --git a/backend/app/api/ws/control.py b/backend/app/api/ws/control.py index 801f81f..eb1ab26 100644 --- a/backend/app/api/ws/control.py +++ b/backend/app/api/ws/control.py @@ -17,7 +17,7 @@ from uuid import UUID from fastapi import APIRouter, WebSocket, WebSocketDisconnect from pydantic import TypeAdapter, ValidationError -from app.api.auth import principal_of, websocket_origin_allowed +from app.api.ws.session import close_fenced, run_command, session_socket from app.config import get_settings from app.db.base import get_sessionmaker from app.db.repo import SessionNodeConflict @@ -47,7 +47,8 @@ from app.domain.roles import Role from app.domain.timers import TimerCode from app.scenarios import store from app.session.dds import prepare_queue -from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.access import can_access +from app.session.hub import hub from app.session.finish import end_session, override_score from app.session.state import SessionState, now_utc from app.session.store import LessonIdentity, LessonRequest, NoteAdded, ScoreOverridden @@ -424,29 +425,14 @@ async def _command(session_id: UUID, event, who) -> None: ) -async def _reject(ws: WebSocket, message: str) -> None: - """Отказ до входа в цикл: сокет закрывается с объяснением, а не молча.""" - await ws.send_text( - ErrorEvent(code=ErrorKind.FORBIDDEN, message=message).model_dump_json() - ) - await ws.close() - - @router.websocket("/ws/control/{session_id}") async def control(ws: WebSocket, session_id: UUID) -> None: - if not websocket_origin_allowed(ws): - await ws.close(code=1008) - return - if hub.is_lease_fenced(session_id): - await ws.close(code=1012) - return - await ws.accept() - - # Пульт преподавателя: управление занятием доступно только ему. - who = principal_of(ws) - if who is None or who.role not in (Role.INSTRUCTOR,): - await _reject(ws, "Недостаточно прав для этого экрана") + # Пульт преподавателя: управление занятием доступно только ему. Занятия + # до `scenario.start` ещё нет — доступ проверяется на каждой команде. + entered = await session_socket(ws, session_id, (Role.INSTRUCTOR,), require_state=False) + if entered is None: return + who, _state = entered # Чтение живёт дольше тика опроса: wait_for отменял бы его, а отмена после # того, как receive уже забрал кадр, теряет команду или websocket.disconnect. # Команда обрабатывается здесь, одной операцией хранилища. @@ -458,10 +444,7 @@ async def control(ws: WebSocket, session_id: UUID) -> None: done, _ = await asyncio.wait({read}, timeout=_FENCE_POLL_SECONDS) if not done: if hub.is_lease_fenced(session_id): - await ws.send_text(ErrorEvent( - code=ErrorKind.INTERNAL, message=LEASE_FENCED_MESSAGE - ).model_dump_json()) - await ws.close(code=1012) + await close_fenced(ws) return continue payload, read = read.result(), None @@ -482,30 +465,16 @@ async def control(ws: WebSocket, session_id: UUID) -> None: message="Для нового занятия создайте новый номер сессии", )) continue - elif existing_state is None or existing_state.owner_login != who.login: + # Запуск пишется своей транзакцией в `store.open`: до неё занятия нет. + await _command(session_id, event, who) + continue + if existing_state is None or not can_access(who, existing_state): hub.to_observers(session_id, ErrorEvent( code=ErrorKind.FORBIDDEN, message="Это занятие принадлежит другому преподавателю", )) continue - - if event.type == "scenario.start": - # Запуск пишется своей транзакцией в `store.open`: до неё занятия нет. - await _command(session_id, event, who) - continue - try: - async with hub.operation(session_id): - await _command(session_id, event, who) - except Exception: - if hub.is_lease_fenced(session_id): - await ws.send_text(ErrorEvent( - code=ErrorKind.INTERNAL, message=LEASE_FENCED_MESSAGE - ).model_dump_json()) - await ws.close(code=1012) - return - raise - if hub.is_lease_fenced(session_id): - await ws.close(code=1012) + if not await run_command(ws, session_id, lambda: _command(session_id, event, who)): return except WebSocketDisconnect: return diff --git a/backend/app/api/ws/observe.py b/backend/app/api/ws/observe.py index adac692..3e36e21 100644 --- a/backend/app/api/ws/observe.py +++ b/backend/app/api/ws/observe.py @@ -15,24 +15,13 @@ from uuid import UUID from fastapi import APIRouter, WebSocket, WebSocketDisconnect -from app.domain.events import ErrorEvent, ErrorKind -from app.api.auth import principal_of, websocket_origin_allowed +from app.api.ws.session import pump, session_socket from app.domain.roles import Role -from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.hub import hub router = APIRouter() -async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None: - while True: - event = await queue.get() - await ws.send_text(event.model_dump_json()) - if (isinstance(event, ErrorEvent) and event.code is ErrorKind.INTERNAL - and event.message == LEASE_FENCED_MESSAGE): - await ws.close(code=1012) - return - - async def _wait_for_disconnect(ws: WebSocket) -> None: """Единственное назначение — дождаться разрыва. Содержимое кадров не читается и никуда не передаётся.""" @@ -42,54 +31,21 @@ async def _wait_for_disconnect(ws: WebSocket) -> None: return -async def _reject(ws: WebSocket, message: str) -> None: - """Отказ до входа в цикл: сокет закрывается с объяснением, а не молча.""" - await ws.send_text( - ErrorEvent(code=ErrorKind.FORBIDDEN, message=message).model_dump_json() - ) - await ws.close() - - @router.websocket("/ws/observe/{session_id}") async def observe(ws: WebSocket, session_id: UUID) -> None: - if not websocket_origin_allowed(ws): - await ws.close(code=1008) - return - if hub.is_lease_fenced(session_id): - await ws.close(code=1012) - return - await ws.accept() - - # Наблюдение за чужим занятием — не для обучающегося. - who = principal_of(ws) - if who is None or who.role not in (Role.INSTRUCTOR, Role.ADMIN): - await _reject(ws, "Недостаточно прав для этого экрана") - return - - state = hub.get(session_id) - if state is None: - await ws.send_text( - ErrorEvent(code=ErrorKind.SESSION_NOT_FOUND, message="Занятие не запущено").model_dump_json() - ) - await ws.close() - return - - # Live state is process-local, so authorize against the owner snapshot on - # the state itself. Instructors may observe only their own sessions; - # administrators retain the cross-owner diagnostic view. - if who.role is Role.INSTRUCTOR and state.owner_login != who.login: - await ws.send_text( - ErrorEvent(code=ErrorKind.SESSION_NOT_FOUND, message="Занятие не запущено").model_dump_json() - ) - await ws.close() + # Наблюдение за чужим занятием — не для обучающегося. Преподаватель видит + # только свои занятия, администратор — любые, для диагностики. + entered = await session_socket(ws, session_id, (Role.INSTRUCTOR, Role.ADMIN)) + if entered is None: return + _who, state = entered # Снимок при подключении обязателен: монитор в классе включают посреди # занятия, и он должен показать текущее состояние, а не ждать событий. await ws.send_text(state.snapshot().model_dump_json()) with hub.observer(session_id) as queue: - sender = asyncio.create_task(_pump(ws, queue)) + sender = asyncio.create_task(pump(ws, queue)) try: await _wait_for_disconnect(ws) except WebSocketDisconnect: diff --git a/backend/app/api/ws/session.py b/backend/app/api/ws/session.py new file mode 100644 index 0000000..fab2b8c --- /dev/null +++ b/backend/app/api/ws/session.py @@ -0,0 +1,160 @@ +"""Общее для сокетов занятия: вход, доставка очереди хаба и выполнение команды. + +Лестница входа одна для всех каналов, чтобы коды закрытия и тексты отказов +не расходились: 1008 — чужой origin, 1012 — занятие ушло другому узлу ещё +до accept, `ErrorEvent` и close — нет прав, нет занятия, не назначено. + +Команда выполняется одной операцией хаба: сверка `(node_id, epoch)` остаётся +в её commit перед каждой мутацией, здесь только место вызова. +""" + +import asyncio +import logging +from collections.abc import Awaitable, Callable, Collection +from uuid import UUID + +from fastapi import WebSocket +from pydantic import BaseModel + +from app.api.auth import Principal, principal_of, websocket_origin_allowed +from app.domain.events import CommandAck, ErrorEvent, ErrorKind +from app.domain.roles import Role +from app.session.access import can_access +from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.state import MAX_STATION_COMMANDS, SessionState + +log = logging.getLogger(__name__) + +NOT_STARTED = "Занятие не запущено" +FORBIDDEN = "Недостаточно прав для этого экрана" +NOT_ASSIGNED = "Занятие не назначено этому обучающемуся" + +#: Сколько ждать, пока `pump` доставит событие fencing сам. Очередь могла +#: переполниться и выпасть из подписки — тогда сокет закрывается здесь. +_FENCE_DELIVERY_SECONDS = 2.0 + + +def _fenced_event() -> ErrorEvent: + return ErrorEvent(code=ErrorKind.INTERNAL, message=LEASE_FENCED_MESSAGE) + + +def _is_fenced_event(item) -> bool: + return (isinstance(item, ErrorEvent) and item.code is ErrorKind.INTERNAL + and item.message == LEASE_FENCED_MESSAGE) + + +async def _refuse(ws: WebSocket, code: ErrorKind, message: str) -> None: + """Отказ до входа в цикл: сокет закрывается с объяснением, а не молча.""" + await ws.send_text(ErrorEvent(code=code, message=message).model_dump_json()) + await ws.close() + + +async def session_socket( + ws: WebSocket, + session_id: UUID, + roles: Collection[Role], + *, + require_state: bool = True, + not_found: str = NOT_STARTED, +) -> tuple[Principal, SessionState | None] | None: + """Принципал и живое занятие — или `None`, если сокет уже закрыт. + + `require_state=False` — для пульта: занятие создаёт его же `scenario.start`, + поэтому вход занятие не смотрит, а доступ проверяется на каждой команде. + """ + if not websocket_origin_allowed(ws): + await ws.close(code=1008) + return None + if hub.is_lease_fenced(session_id): + await ws.close(code=1012) + return None + await ws.accept() + + who = principal_of(ws) + if who is None or who.role not in roles: + await _refuse(ws, ErrorKind.FORBIDDEN, FORBIDDEN) + return None + if not require_state: + return who, None + state = hub.get(session_id) + if state is None: + await _refuse(ws, ErrorKind.SESSION_NOT_FOUND, not_found) + return None + if not can_access(who, state): + # Чужому преподавателю занятие «не существует»: отказ не выдаёт владельца. + if who.role is Role.TRAINEE: + await _refuse(ws, ErrorKind.FORBIDDEN, NOT_ASSIGNED) + else: + await _refuse(ws, ErrorKind.SESSION_NOT_FOUND, not_found) + return None + return who, state + + +async def pump(ws: WebSocket, queue: asyncio.Queue) -> None: + """Очередь хаба в сокет; событие fencing закрывает его с 1012.""" + while True: + item = await queue.get() + # Бинарь — звук звонящего, без обёртки JSON (docs/arch/CONTRACT.md). + if isinstance(item, bytes): + await ws.send_bytes(item) + continue + await ws.send_text(item.model_dump_json()) + if _is_fenced_event(item): + await ws.close(code=1012) + return + + +async def close_fenced(ws: WebSocket, sender: asyncio.Task | None = None) -> None: + """Структурное событие fencing и 1012 вместо безымянного 1006. + + С очередью событие уже разослал `hub.fence` — его доставляет `pump`. + Без очереди (пульт) сообщение уходит прямо в сокет. + """ + if sender is not None: + done, _ = await asyncio.wait({sender}, timeout=_FENCE_DELIVERY_SECONDS) + if done: + return + sender.cancel() + await ws.send_text(_fenced_event().model_dump_json()) + await ws.close(code=1012) + + +async def run_command( + ws: WebSocket, + session_id: UUID, + handler: Callable[[], Awaitable[None]], + *, + command_id: UUID | None = None, + ack: Callable[[BaseModel], None] | None = None, + sender: asyncio.Task | None = None, +) -> bool: + """Одна команда — одна операция хаба. `False` — сокет закрыт fencing. + + С `command_id` повтор уже зафиксированной команды только подтверждается: + id входит в тот же снимок, что и переход, а `CommandAck` копится в + операции и уходит после её commit. + """ + state = hub.get(session_id) + if command_id is not None and state is not None: + if str(command_id) in state.processed_station_commands: + if ack is not None: + ack(CommandAck(command_id=command_id)) + return True + try: + async with hub.operation(session_id): + await handler() + if command_id is not None and state is not None: + state.processed_station_commands.append(str(command_id)) + del state.processed_station_commands[:-MAX_STATION_COMMANDS] + if ack is not None: + ack(CommandAck(command_id=command_id)) + except Exception: + if not hub.is_lease_fenced(session_id): + raise + log.info("закрытие WebSocket после fencing занятия %s", session_id) + await close_fenced(ws, sender) + return False + if hub.is_lease_fenced(session_id): + await close_fenced(ws, sender) + return False + return True diff --git a/backend/app/api/ws/station.py b/backend/app/api/ws/station.py index b6e9aa3..8481c09 100644 --- a/backend/app/api/ws/station.py +++ b/backend/app/api/ws/station.py @@ -7,21 +7,19 @@ этаж, и отбивает карточку обратно. Неполнота КИО перестаёт быть процентом в отчёте и становится сорванным выездом с конкретной причиной. -Правила пульта — в `DdsDesk.apply`; здесь разбор команды, отсев повторов, -операция хранилища и рассылка итога. +Правила пульта — в `DdsDesk.apply`; здесь разбор команды и рассылка итога. +Отсев повторов и операция хранилища — в `run_command`. """ import asyncio -import logging from uuid import UUID from fastapi import APIRouter, WebSocket, WebSocketDisconnect from pydantic import TypeAdapter, ValidationError -from app.api.auth import principal_of, websocket_origin_allowed +from app.api.ws.session import pump, run_command, session_socket from app.domain.events import ( CallEndReason, - CommandAck, ErrorEvent, ErrorKind, StationState, @@ -29,9 +27,8 @@ from app.domain.events import ( ) from app.domain.roles import Role from app.session.finish import end_session -from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.hub import hub -log = logging.getLogger(__name__) router = APIRouter() _adapter = TypeAdapter(StationToServer) @@ -56,61 +53,16 @@ async def _handle(session_id: UUID, state, event) -> None: hub.to_observers(session_id, state.snapshot()) -async def _pump(ws: WebSocket, queue: asyncio.Queue) -> None: - while True: - event = await queue.get() - await ws.send_text(event.model_dump_json()) - if (isinstance(event, ErrorEvent) and event.code is ErrorKind.INTERNAL - and event.message == LEASE_FENCED_MESSAGE): - await ws.close(code=1012) - return - - -async def _reject(ws: WebSocket, message: str) -> None: - """Отказ до входа в цикл: сокет закрывается с объяснением, а не молча.""" - await ws.send_text( - ErrorEvent(code=ErrorKind.FORBIDDEN, message=message).model_dump_json() - ) - await ws.close() - - @router.websocket("/ws/station/{session_id}") async def station(ws: WebSocket, session_id: UUID, role: str = "dds") -> None: - if not websocket_origin_allowed(ws): - await ws.close(code=1008) - return - if hub.is_lease_fenced(session_id): - await ws.close(code=1012) - return - await ws.accept() - # За АРМ ДДС садится обучающийся, преподаватель смотрит и подменяет. - who = principal_of(ws) - if who is None or who.role not in (Role.INSTRUCTOR, Role.TRAINEE): - await _reject(ws, "Недостаточно прав для этого экрана") - return - - state = hub.get(session_id) - if state is None: - await ws.send_text( - ErrorEvent(code=ErrorKind.SESSION_NOT_FOUND, message="Занятие не запущено").model_dump_json() - ) - await ws.close() - return - if who.role is Role.INSTRUCTOR and state.owner_login != who.login: - await ws.send_text( - ErrorEvent(code=ErrorKind.SESSION_NOT_FOUND, message="Занятие не запущено").model_dump_json() - ) - await ws.close() - return - if who.role is Role.TRAINEE and ( - state.trainee_id is None or state.trainee_id != who.trainee_id - ): - await _reject(ws, "Занятие не назначено этому обучающемуся") + entered = await session_socket(ws, session_id, (Role.INSTRUCTOR, Role.TRAINEE)) + if entered is None: return + _who, state = entered with hub.station(session_id) as queue: - sender = asyncio.create_task(_pump(ws, queue)) + sender = asyncio.create_task(pump(ws, queue)) try: # Карточка, переданная до подключения станции, не теряется: # диспетчер садится за АРМ, когда вызов уже идёт. @@ -137,32 +89,13 @@ async def station(ws: WebSocket, session_id: UUID, role: str = "dds") -> None: message="Некорректный идентификатор команды"), ) continue - if command_id is not None and str(command_id) in state.processed_station_commands: - # The checkpoint already proves this exact command committed. - # Re-ack it without rerunning its business transition. - hub.to_station(session_id, CommandAck(command_id=command_id)) - continue - async with hub.operation(session_id): - await _handle(session_id, state, event) - if command_id is not None: - # ID команды входит в тот же снимок, что и её переход: - # подтверждение уходит только после их общего commit. - state.processed_station_commands.append(str(command_id)) - del state.processed_station_commands[:-512] - hub.to_station(session_id, CommandAck(command_id=command_id)) + if not await run_command( + ws, session_id, lambda: _handle(session_id, state, event), + command_id=command_id, sender=sender, + ack=lambda item: hub.to_station(session_id, item), + ): + return except WebSocketDisconnect: return - except Exception: # noqa: BLE001 — failed durable transition may fence the owner - if not state.lease_fenced: - raise - log.info( - "закрытие станционного WebSocket после fencing занятия %s", - session_id, - ) - # `hub.operation` broadcasts a structured fence event before - # propagating the failed database write. Let the sender deliver - # that event and close with 1012 instead of an opaque 1006. - await asyncio.gather(sender, return_exceptions=True) - return finally: sender.cancel() diff --git a/backend/app/session/access.py b/backend/app/session/access.py new file mode 100644 index 0000000..fc9a751 --- /dev/null +++ b/backend/app/session/access.py @@ -0,0 +1,30 @@ +"""Кто допущен к занятию — одно правило для сокетов и HTTP. + +Живое состояние и строка сессии в БД несут одни и те же `owner_login` и +`trainee_id`, поэтому правило принимает любое из них. Владение группами и +сценариями — другое понятие и сюда не входит. +""" + +from typing import Protocol +from uuid import UUID + +from app.domain.roles import Role + + +class _Lesson(Protocol): + owner_login: str | None + trainee_id: UUID | None + + +def can_access(who, lesson: _Lesson) -> bool: + """Преподаватель — только своё занятие, курсант — только назначенное ему. + + Занятие без курсанта не открывается ни одному курсанту: иначе `None == None` + пустил бы учётку без профиля. Роль экрана проверяет вызывающий; администратору + занятие не закрыто — ему доступна диагностика, а не оценки. + """ + if who.role is Role.INSTRUCTOR: + return lesson.owner_login == who.login + if who.role is Role.TRAINEE: + return lesson.trainee_id is not None and lesson.trainee_id == who.trainee_id + return True diff --git a/backend/tests/test_db.py b/backend/tests/test_db.py index 04f442a..2eea63f 100644 --- a/backend/tests/test_db.py +++ b/backend/tests/test_db.py @@ -18,7 +18,7 @@ from app.api.http import ( groups as groups_api, materials as materials_api, scenario_submissions, scenarios as scenarios_api, sessions as sessions_api, ) -from app.api.ws import control as control_ws, station as station_ws +from app.api.ws import session as session_ws from app.api.http.scenarios import _hidden_scenario_ids from app.config import get_settings from app.db import repo @@ -881,8 +881,10 @@ async def test_kio_submission_moderation_and_dds_card_persist_on_real_postgres( scenarios_api, "require", lambda *_args, **_kwargs: instructor_principal, ) - monkeypatch.setattr(control_ws, "principal_of", lambda _ws: instructor_principal) - monkeypatch.setattr(station_ws, "principal_of", lambda _ws: trainee_principal) + monkeypatch.setattr( + session_ws, "principal_of", + lambda ws: instructor_principal if "/ws/control/" in ws.scope["path"] else trainee_principal, + ) scenario_id = None live_session_id = uuid4() diff --git a/backend/tests/test_route_auth_contract.py b/backend/tests/test_route_auth_contract.py index 05ad17a..5c9e5d0 100644 --- a/backend/tests/test_route_auth_contract.py +++ b/backend/tests/test_route_auth_contract.py @@ -199,12 +199,22 @@ def test_every_websocket_route_checks_a_principal_before_serving(): route = _route_declaration(node) if route is None or route[0] != "websocket": continue - if "principal_of" not in _called_names(node): + if not _called_names(node) & {"principal_of", "session_socket"}: missing.append(f"{source.name}:{node.name} ({route[1]})") assert not missing, "WebSocket routes without principal check: " + "; ".join(missing) +def test_session_socket_entry_checks_a_principal(): + """Сокеты занятия делегируют проверку принципала общему входу.""" + tree = ast.parse((API_ROOT / "ws" / "session.py").read_text(encoding="utf-8")) + entry = next( + node for node in tree.body + if isinstance(node, ast.AsyncFunctionDef) and node.name == "session_socket" + ) + assert "principal_of" in _called_names(entry) + + def test_http_routes_match_the_reviewed_role_gate_matrix(): found: dict[tuple[str, str, str], tuple[tuple[str, ...], ...]] = {} session_guards: set[tuple[str, str, str]] = set() diff --git a/backend/tests/test_session_access.py b/backend/tests/test_session_access.py index b8d4fbf..866af53 100644 --- a/backend/tests/test_session_access.py +++ b/backend/tests/test_session_access.py @@ -14,6 +14,7 @@ from app.api.auth import Principal from app.api.http import sessions from app.api.ws import call as call_ws from app.api.ws import observe as observe_ws +from app.api.ws import session as session_ws from app.api.ws import station as station_ws from app.domain.events import Exercise, SessionMode, Speaker from app.domain.roles import Role @@ -306,8 +307,8 @@ async def test_instructor_cannot_join_foreign_live_session(monkeypatch, ws_modul session_id = uuid4() who = Principal(login="teacher-a", full_name="Преподаватель A", role=Role.INSTRUCTOR) state = SimpleNamespace(owner_login="teacher-b") - monkeypatch.setattr(ws_module, "websocket_origin_allowed", lambda _ws: True) - monkeypatch.setattr(ws_module, "principal_of", lambda _ws: who) + monkeypatch.setattr(session_ws, "websocket_origin_allowed", lambda _ws: True) + monkeypatch.setattr(session_ws, "principal_of", lambda _ws: who) monkeypatch.setattr(ws_module.hub, "get", lambda _session_id: state) class Socket: diff --git a/backend/tests/test_session_socket.py b/backend/tests/test_session_socket.py new file mode 100644 index 0000000..c2951d6 --- /dev/null +++ b/backend/tests/test_session_socket.py @@ -0,0 +1,312 @@ +"""Вход в сокет занятия и выполнение команды: права и fencing одного модуля.""" + +import asyncio +import json +from uuid import uuid4 + +import pytest + +from app.api.auth import Principal +from app.api.ws import call as call_ws +from app.api.ws import control as control_ws +from app.api.ws import session as session_ws +from app.api.ws import station as station_ws +from app.domain.events import CommandAck, Exercise, SessionMode, StationState +from app.domain.roles import Role +from app.session.access import can_access +from app.session.hub import LEASE_FENCED_MESSAGE, hub +from app.session.state import SessionState +from app.session.store import MemorySessionStore + +TEACHER = Principal(login="teacher-a", full_name="Преподаватель A", role=Role.INSTRUCTOR) +STRANGER = Principal(login="teacher-b", full_name="Преподаватель B", role=Role.INSTRUCTOR) +ADMIN = Principal(login="root", full_name="Администратор", role=Role.ADMIN) + + +def trainee(trainee_id=None) -> Principal: + return Principal(login="cadet", full_name="Курсант", role=Role.TRAINEE, + trainee_id=trainee_id or uuid4()) + + +def lesson_state(**fields) -> SessionState: + return SessionState( + session_id=uuid4(), scenario_id="test", scenario_title="Тест", level="L1", + mode=SessionMode.TRAINING, owner_login=TEACHER.login, **fields, + ) + + +class Socket: + """Сокет без сети: что отправлено и с каким кодом закрыт.""" + + def __init__(self, incoming=()): + self.accepted = False + self.closed = False + self.close_code = None + self.sent: list[dict] = [] + self.incoming = list(incoming) + + async def accept(self): + self.accepted = True + + async def send_text(self, message): + self.sent.append(json.loads(message)) + + async def send_bytes(self, _data): + pass + + async def close(self, code=None): + self.closed = True + self.close_code = code + + async def receive_json(self): + if self.incoming: + return self.incoming.pop(0) + await asyncio.sleep(5) + + async def receive(self): + if self.incoming: + return {"type": "websocket.receive", "text": json.dumps(self.incoming.pop(0))} + await asyncio.sleep(5) + return {"type": "websocket.disconnect"} + + +class BrokenStore(MemorySessionStore): + async def commit(self, _state, _records=()): + raise OSError("simulated database partition") + + +@pytest.fixture +def lesson(monkeypatch): + """Занятие в общем хабе и принципал, которого подставляет тест.""" + monkeypatch.setattr(session_ws, "websocket_origin_allowed", lambda _ws: True) + current = {"who": TEACHER} + monkeypatch.setattr(session_ws, "principal_of", lambda _ws: current["who"]) + old_store = hub.store + hub.store = MemorySessionStore() + states = [] + + def make(who=TEACHER, **fields): + current["who"] = who + state = lesson_state(**fields) + hub.register(state) + states.append(state) + return state + + make.set_who = lambda who: current.update(who=who) + yield make + hub.store = old_store + for state in states: + hub.drop(state.session_id) + + +def enter(socket, session_id, roles=(Role.INSTRUCTOR, Role.TRAINEE), **options): + return asyncio.run(session_ws.session_socket(socket, session_id, roles, **options)) + + +# ── предикат доступа ── + + +def test_access_is_owner_instructor_or_assigned_trainee(): + cadet = trainee() + state = lesson_state(trainee_id=cadet.trainee_id) + assert can_access(TEACHER, state) + assert not can_access(STRANGER, state) + assert can_access(cadet, state) + assert not can_access(trainee(), state) + assert can_access(ADMIN, state), "администратор видит занятие для диагностики" + + +def test_unassigned_session_is_closed_for_every_trainee(): + assert not can_access(Principal(login="x", full_name="x", role=Role.TRAINEE), lesson_state()) + + +# ── вход в сокет ── + + +def test_foreign_origin_is_closed_with_1008_before_accept(lesson, monkeypatch): + state = lesson() + monkeypatch.setattr(session_ws, "websocket_origin_allowed", lambda _ws: False) + socket = Socket() + assert enter(socket, state.session_id) is None + assert not socket.accepted and socket.close_code == 1008 + + +def test_fenced_session_is_closed_with_1012_before_accept(lesson): + state = lesson() + state.lease_fenced = True + socket = Socket() + assert enter(socket, state.session_id) is None + assert not socket.accepted and socket.close_code == 1012 + + +@pytest.mark.parametrize("who", [None, ADMIN]) +def test_missing_principal_or_wrong_role_is_forbidden(lesson, who): + state = lesson() + lesson.set_who(who) + socket = Socket() + assert enter(socket, state.session_id) is None + assert socket.accepted and socket.closed + assert [event["code"] for event in socket.sent] == ["forbidden"] + + +def test_missing_session_is_not_found(lesson): + lesson() + socket = Socket() + assert enter(socket, uuid4()) is None + assert socket.closed + assert [event["code"] for event in socket.sent] == ["session_not_found"] + + +def test_foreign_instructor_sees_not_found_without_owner_login(lesson): + state = lesson() + lesson.set_who(STRANGER) + socket = Socket() + assert enter(socket, state.session_id) is None + assert socket.closed + assert [event["code"] for event in socket.sent] == ["session_not_found"] + assert TEACHER.login not in json.dumps(socket.sent, ensure_ascii=False) + + +@pytest.mark.parametrize("assigned", [True, False]) +def test_unassigned_trainee_is_forbidden(lesson, assigned): + state = lesson(trainee_id=uuid4() if assigned else None) + lesson.set_who(trainee()) + socket = Socket() + assert enter(socket, state.session_id) is None + assert socket.closed + assert [event["code"] for event in socket.sent] == ["forbidden"] + + +def test_owner_and_assigned_trainee_enter(lesson): + cadet = trainee() + state = lesson(trainee_id=cadet.trainee_id) + for who in (TEACHER, cadet): + lesson.set_who(who) + socket = Socket() + assert enter(socket, state.session_id) == (who, state) + assert socket.accepted and not socket.closed and socket.sent == [] + + +def test_optional_session_lets_control_in_before_start(lesson): + lesson() + socket = Socket() + session_id = uuid4() + assert enter(socket, session_id, (Role.INSTRUCTOR,), require_state=False) == (TEACHER, None) + assert socket.accepted and not socket.closed + + +# ── выполнение команды ── + + +def test_repeated_command_is_acked_without_second_transition(lesson): + state = lesson() + command_id = uuid4() + calls = [] + + async def scenario(): + with hub.station(state.session_id) as queue: + async def handler(): + calls.append(1) + for _ in range(2): + assert await session_ws.run_command( + Socket(), state.session_id, handler, command_id=command_id, + ack=lambda event: hub.to_station(state.session_id, event), + ) + return [queue.get_nowait() for _ in range(queue.qsize())] + + events = asyncio.run(scenario()) + assert calls == [1] + assert events == [CommandAck(command_id=command_id)] * 2 + assert str(command_id) in state.processed_station_commands + + +def test_ack_is_sent_only_after_commit(lesson): + state = lesson() + command_id = uuid4() + seen_at_commit = [] + + class WatchingStore(MemorySessionStore): + queue: asyncio.Queue + + async def commit(self, state, records=()): + seen_at_commit.append( + (self.queue.qsize(), str(command_id) in state.processed_station_commands) + ) + + watching = hub.store = WatchingStore() + + async def handler(): + hub.to_station(state.session_id, StationState(snapshot=state.station_snapshot())) + + async def scenario(): + with hub.station(state.session_id) as queue: + watching.queue = queue + await session_ws.run_command( + Socket(), state.session_id, handler, command_id=command_id, + ack=lambda event: hub.to_station(state.session_id, event), + ) + return [queue.get_nowait() for _ in range(queue.qsize())] + + events = asyncio.run(scenario()) + assert seen_at_commit == [(0, True)], "до commit клиенту ничего не ушло, id уже в снимке" + assert events[-1] == CommandAck(command_id=command_id) + + +def test_fencing_without_queue_is_reported_on_the_socket(lesson): + state = lesson() + hub.store = BrokenStore() + socket = Socket() + + async def handler(): + pass + + assert not asyncio.run(session_ws.run_command(socket, state.session_id, handler)) + assert state.lease_fenced + assert socket.sent[-1] == {"type": "error", "code": "internal", "message": LEASE_FENCED_MESSAGE} + assert socket.close_code == 1012 + + +def test_error_without_fencing_propagates(): + """Занятия нет на узле — fencing не было, исключение уходит как раньше.""" + async def handler(): + raise ValueError("bug") + + with pytest.raises(ValueError): + asyncio.run(session_ws.run_command(Socket(), uuid4(), handler)) + + +# ── fencing на каналах: ErrorEvent и 1012, а не 1006 ── + + +def _fenced_close(socket): + assert socket.close_code == 1012 + fenced = [event for event in socket.sent if event.get("type") == "error"] + assert fenced[-1] == {"type": "error", "code": "internal", "message": LEASE_FENCED_MESSAGE} + + +def test_station_command_fencing_closes_with_1012(lesson): + state = lesson(exercise=Exercise.DDS) + hub.store = BrokenStore() + socket = Socket([{"type": "card.ack", "comment": "принято", + "_command_id": str(uuid4())}]) + asyncio.run(asyncio.wait_for(station_ws.station(socket, state.session_id), 3)) + assert state.lease_fenced + _fenced_close(socket) + + +def test_call_command_fencing_closes_with_1012(lesson): + state = lesson(exercise=Exercise.CALL) + hub.store = BrokenStore() + socket = Socket([{"type": "callback.dial"}]) + asyncio.run(asyncio.wait_for(call_ws.call(socket, state.session_id), 3)) + assert state.lease_fenced + _fenced_close(socket) + + +def test_control_command_fencing_closes_with_1012(lesson): + state = lesson() + hub.store = BrokenStore() + socket = Socket([{"type": "reference.play"}]) + asyncio.run(asyncio.wait_for(control_ws.control(socket, state.session_id), 3)) + assert state.lease_fenced + _fenced_close(socket) diff --git a/backend/tests/test_ws_ownership.py b/backend/tests/test_ws_ownership.py index 1c00a24..9c936d0 100644 --- a/backend/tests/test_ws_ownership.py +++ b/backend/tests/test_ws_ownership.py @@ -10,6 +10,7 @@ import pytest from fastapi.testclient import TestClient from app.api.auth import Principal +from app.api.ws import session as session_ws from app.domain.roles import Role from app.main import app from app.scenarios import store @@ -47,7 +48,7 @@ def test_trainee_cannot_open_foreign_or_unassigned_call_or_station(client, monke for role_name, module in modules.items(): who = Principal(login="foreign", full_name="Чужой", role=Role.TRAINEE, trainee_id=foreign_id) - monkeypatch.setattr(module, "principal_of", lambda _ws, user=who: user) + monkeypatch.setattr(session_ws, "principal_of", lambda _ws, user=who: user) with client.websocket_connect(f"/ws/{role_name}/{session_id}") as socket: event = socket.receive_json() assert event["type"] == "error" and event["code"] == "forbidden" @@ -56,7 +57,7 @@ def test_trainee_cannot_open_foreign_or_unassigned_call_or_station(client, monke for role_name, module in modules.items(): who = Principal(login="owner", full_name="Назначенный", role=Role.TRAINEE, trainee_id=owner_id) - monkeypatch.setattr(module, "principal_of", lambda _ws, user=who: user) + monkeypatch.setattr(session_ws, "principal_of", lambda _ws, user=who: user) with client.websocket_connect(f"/ws/{role_name}/{session_id}") as socket: socket.send_json({"type": "unknown"}) for _ in range(8): @@ -114,8 +115,8 @@ def test_idle_control_socket_closes_when_backend_lease_becomes_uncertain(monkeyp ) hub.register(state) who = Principal(login="lease-owner", full_name="Преподаватель", role=Role.INSTRUCTOR) - monkeypatch.setattr(control_module, "websocket_origin_allowed", lambda _ws: True) - monkeypatch.setattr(control_module, "principal_of", lambda _ws: who) + monkeypatch.setattr(session_ws, "websocket_origin_allowed", lambda _ws: True) + monkeypatch.setattr(session_ws, "principal_of", lambda _ws: who) class IdleSocket: accepted = False @@ -180,8 +181,8 @@ def test_control_command_checkpoint_failure_returns_fencing_error(monkeypatch): old_store = hub.store hub.store = BrokenStore() hub.register(state) - monkeypatch.setattr(control_module, "websocket_origin_allowed", lambda _ws: True) - monkeypatch.setattr(control_module, "principal_of", lambda _ws: who) + monkeypatch.setattr(session_ws, "websocket_origin_allowed", lambda _ws: True) + monkeypatch.setattr(session_ws, "principal_of", lambda _ws: who) try: with hub.observer(session_id) as observer_queue: asyncio.run(control_module.control(socket, session_id)) @@ -214,8 +215,8 @@ def test_control_does_not_lose_command_read_during_fencing_poll(monkeypatch): ) hub.register(state) who = Principal(login="lease-owner", full_name="Преподаватель", role=Role.INSTRUCTOR) - monkeypatch.setattr(control_module, "websocket_origin_allowed", lambda _ws: True) - monkeypatch.setattr(control_module, "principal_of", lambda _ws: who) + monkeypatch.setattr(session_ws, "websocket_origin_allowed", lambda _ws: True) + monkeypatch.setattr(session_ws, "principal_of", lambda _ws: who) class SlowReturnSocket: def __init__(self):