Harden WebSocket message handling and stop logging secrets

docs/pages/user_config.md

@@ -76,13 +76,19 @@ Once users have been created, their permissions can be configured by clicking th
 
 RBAC configuration determines what shows a user can access and what actions they can perform within those shows.
 
+### Log Redaction
+
+The **Enable Log Redaction** system setting is on by default. While it is on, passwords, tokens and other credentials are replaced with a placeholder in logged HTTP request bodies, and WebSocket messages are logged as their operation name and size only.
+
+Turn it off to see the raw HTTP and WebSocket content while debugging. Request bodies and WebSocket messages are logged at debug level, so also set **Log Level** to `DEBUG` (the default is `INFO`). **Warning:** with redaction off, the logs contain credentials and authentication tokens, so only disable it temporarily and treat the log files as sensitive. A value you have saved is never changed by a later change to a setting's default.
+
 ### Log Viewer
 
 The **Logs** tab (admin only) provides a real-time view of server and client log entries stored in the in-memory log buffer.
 
 #### Sources
 
-- **Server** — Application-level logs from the DigiScript backend: HTTP request access logs, request body debug logs, WebSocket connection events, and general application messages.
+- **Server** — Application-level logs from the DigiScript backend: HTTP request access logs, request body and WebSocket message debug logs (only shown when the Log Level is `DEBUG`), WebSocket connection events, and general application messages.
 - **Client** — Logs forwarded from connected browser clients via the `/api/v1/logs/batch` endpoint.
 
 #### Username Attribution

server/controllers/ws_controller.py

@@ -132,8 +132,12 @@ def update_session(self, user_id=None) -> bool:
                         user_id=user_id,
                     )
                 )
-            if self.current_user_id:
-                user = session.get(User, self.current_user_id)
+            user = (
+                session.get(User, self.current_user_id)
+                if self.current_user_id
+                else None
+            )
+            if user is not None:
                 user.last_seen = datetime.datetime.now(tz=datetime.timezone.utc)
             session.commit()
         return cleared
@@ -467,17 +471,34 @@ async def on_message(self, message: Union[str, bytes]):
             if self.current_username
             else self.request.remote_ip
         )
-        get_logger().debug(f"WebSocket message from {user_part}: {message}")
+        raw = message
+        size = len(message)
 
         try:
             message = json.loads(message)
             ws_op = message["OP"]
-        except (json.JSONDecodeError, KeyError) as e:
+            if not isinstance(ws_op, str):
+                raise TypeError("OP must be a string")
+        except (json.JSONDecodeError, KeyError, TypeError) as e:
+            if self.application.digi_settings.get_sync("log_redaction"):
+                content = f"<{size} bytes>"
+            else:
+                content = raw
             get_logger().warning(
-                f"Malformed WS message from {self.request.remote_ip}: {e}"
+                f"Malformed WS message from {self.request.remote_ip}: "
+                f"{type(e).__name__}: {e} {content}"
             )
             return
 
+        # With redaction on never log the payload: AUTHENTICATE / REFRESH_TOKEN
+        # carry live JWTs.
+        if self.application.digi_settings.get_sync("log_redaction"):
+            get_logger().debug(
+                f"WebSocket message from {user_part}: {ws_op} ({size} bytes)"
+            )
+        else:
+            get_logger().debug(f"WebSocket message from {user_part}: {raw}")
+
         # Connection boundary: an exception escaping on_message ends Tornado's
         # read loop without calling on_close, leaving a handler that no longer
         # reads frames but still counts as a connected owner. Each op keeps its
@@ -529,43 +550,58 @@ async def _handle_session_op(self, ws_op: str, message: dict) -> None:
             elif ws_op == "REFRESH_CLIENT":
                 self._resume_client(message.get("DATA"))
             elif ws_op in self._LEADER_OP_HANDLERS:
-                show_session = (
-                    session.get(ShowSession, show.current_session_id)
-                    if show and show.current_session_id
-                    else None
-                )
-                # Ownership rule: only the authenticated show user on its own row.
-                if self._is_leader(session, show_session):
-                    handler = getattr(self, self._LEADER_OP_HANDLERS[ws_op])
-                    await handler(session, show_session, entry, message["DATA"])
-                else:
-                    get_logger().debug(
-                        f"Ignored {ws_op} from client "
-                        f"{self.__getattribute__('internal_id')} (user "
-                        f"{self.current_user_id}): not the live-show leader"
-                    )
+                await self._handle_leader_op(session, show, entry, ws_op, message)
             elif ws_op == "LIVE_SHOW_JUMP_TO_PAGE":
-                show_session = (
-                    session.get(ShowSession, show.current_session_id)
-                    if show and show.current_session_id
-                    else None
-                )
-                if show_session and self._user_has_show_execute_access(session, show):
-                    await self._op_jump_to_page(
-                        session, show_session, entry, message["DATA"]
-                    )
-                else:
-                    get_logger().debug(
-                        f"Ignored {ws_op} from client "
-                        f"{self.__getattribute__('internal_id')} (user "
-                        f"{self.current_user_id}): no show execute access"
-                    )
+                await self._handle_jump_op(session, show, entry, ws_op, message)
             else:
                 get_logger().warning(
                     f"Unknown OP {ws_op} received from "
                     f"WebSocket connection {self.request.remote_ip}"
                 )
 
+    @staticmethod
+    def _live_show_session(session, show: Optional[Show]) -> Optional[ShowSession]:
+        """Return the show's current live ShowSession, or None."""
+        if show and show.current_session_id:
+            return session.get(ShowSession, show.current_session_id)
+        return None
+
+    async def _handle_leader_op(
+        self, session, show: Optional[Show], entry, ws_op: str, message: dict
+    ) -> None:
+        """Run a leader-only live-show op if this connection is the leader."""
+        show_session = self._live_show_session(session, show)
+        # Ownership rule: only the authenticated show user on its own row.
+        if not self._is_leader(session, show_session):
+            get_logger().debug(
+                f"Ignored {ws_op} from client "
+                f"{self.__getattribute__('internal_id')} (user "
+                f"{self.current_user_id}): not the live-show leader"
+            )
+            return
+        data = message.get("DATA")
+        if not isinstance(data, dict):
+            get_logger().warning(f"Ignored {ws_op}: DATA must be an object")
+            return
+        handler = getattr(self, self._LEADER_OP_HANDLERS[ws_op])
+        await handler(session, show_session, entry, data)
+
+    async def _handle_jump_op(
+        self, session, show: Optional[Show], entry, ws_op: str, message: dict
+    ) -> None:
+        """Run ``LIVE_SHOW_JUMP_TO_PAGE`` if the user has show EXECUTE access."""
+        show_session = self._live_show_session(session, show)
+        if show_session and self._user_has_show_execute_access(session, show):
+            await self._op_jump_to_page(
+                session, show_session, entry, message.get("DATA")
+            )
+            return
+        get_logger().debug(
+            f"Ignored {ws_op} from client "
+            f"{self.__getattribute__('internal_id')} (user "
+            f"{self.current_user_id}): no show execute access"
+        )
+
     def _resume_client(self, new_uuid: Any) -> None:
         """Handle ``REFRESH_CLIENT``: a reconnecting page resumes its old uuid.
 
@@ -638,19 +674,39 @@ def _resume_client(self, new_uuid: Any) -> None:
 
     async def _op_script_scroll(self, session, show_session, entry, data) -> None:
         """Leader op ``SCRIPT_SCROLL``: record and broadcast the leader's line."""
-        show_session.latest_line_ref = data["current_line"]
+        line_ref = data.get("current_line")
+        if not isinstance(line_ref, str):
+            get_logger().warning("Ignored SCRIPT_SCROLL: invalid current_line")
+            return
+        show_session.latest_line_ref = line_ref
         session.commit()
         await self.application.ws_send_to_all("NOOP", "SCRIPT_SCROLL", data)
 
     async def _op_begin_interval(self, session, show_session, entry, data) -> None:
         """Leader op ``BEGIN_INTERVAL``: start an interval after the given act."""
-        act: Act = session.get(Act, data["actId"])
+        act_id = data.get("actId")
+        length = data.get("length")
         if not entry:
+            get_logger().warning(
+                "Ignored BEGIN_INTERVAL: no session entry for this connection"
+            )
+            return
+        if (
+            isinstance(act_id, bool)
+            or not isinstance(act_id, int)
+            or isinstance(length, bool)
+            or not isinstance(length, (int, float))
+        ):
+            get_logger().warning("Ignored BEGIN_INTERVAL: invalid actId or length")
+            return
+        act: Optional[Act] = session.get(Act, act_id)
+        if act is None:
+            get_logger().warning("Ignored BEGIN_INTERVAL: unknown act")
             return
         show_interval = Interval(
             session_id=show_session.id,
             act_id=act.id,
-            initial_length=data["length"],
+            initial_length=length,
         )
         session.add(show_interval)
         session.flush()
@@ -826,7 +882,10 @@ async def _handle_script_room_op(self, ws_op: str, message: dict):
             get_logger().warning("RoomManager not available, ignoring collab OP")
             return
 
-        data = message.get("DATA", {})
+        data = message.get("DATA")
+        if not isinstance(data, dict):
+            get_logger().warning(f"{ws_op}: DATA is not an object; treating as empty")
+            data = {}
 
         # Everything that touches the shared draft only exists in collaborative mode;
         # in classic mode a stray client must not be able to open a room.

server/digi_server/settings.py

@@ -265,7 +265,7 @@ def init_settings(self):
         self.define(
             "log_level",
             str,
-            "DEBUG",
+            "INFO",
             True,
             self._application.regen_logging,
             display_name="Log Level",
@@ -302,10 +302,14 @@ def init_settings(self):
         self.define(
             "log_redaction",
             bool,
-            False,
+            True,
             True,
             display_name="Enable Log Redaction",
-            help_text="When enabled, potentially sensitive information will be redacted from logs.",
+            help_text=(
+                "When enabled (the default), passwords, tokens and WebSocket message "
+                "contents are kept out of the logs. Disable only while debugging: "
+                "logs will then contain credentials and tokens."
+            ),
             category="Logging",
         )
         self.define(

server/test/controllers/test_ws_controller.py

@@ -3475,6 +3475,102 @@ async def test_jump_to_page_without_page_does_not_kill_connection(self):
         await self._still_responsive(ws_l, self.admin_id)
         ws_l.close()
 
+    @gen_test
+    async def test_leader_ops_with_malformed_data_are_ignored(self):
+        """Non-object / wrong-typed DATA on leader ops is rejected without
+        changing state or killing the connection.
+        """
+        ws_l, uuid_l = await self._connect_and_auth(self.admin_id)
+        self._start_live_session(uuid_l, self.admin_id)
+
+        def latest():
+            with self._app.get_db().sessionmaker() as session:
+                show = session.get(Show, self.show_id)
+                live = session.get(ShowSession, show.current_session_id)
+                return live.latest_line_ref
+
+        before = latest()
+        for op in ("SCRIPT_SCROLL", "BEGIN_INTERVAL", "END_INTERVAL"):
+            for bad in (None, [], "x", 5):
+                await ws_l.write_message(json.dumps({"OP": op, "DATA": bad}))
+            await ws_l.write_message(json.dumps({"OP": op}))
+        await ws_l.write_message(
+            json.dumps({"OP": "SCRIPT_SCROLL", "DATA": {"current_line": ["x"]}})
+        )
+        for bad_interval in (
+            {"actId": "a", "length": 1},
+            {"actId": True, "length": 1},
+            {"actId": 1, "length": True},
+        ):
+            await ws_l.write_message(
+                json.dumps({"OP": "BEGIN_INTERVAL", "DATA": bad_interval})
+            )
+        await ws_l.write_message(json.dumps({"OP": "LIVE_SHOW_JUMP_TO_PAGE"}))
+        await self._still_responsive(ws_l, self.admin_id)
+        self.assertEqual(before, latest())
+        with self._app.get_db().sessionmaker() as session:
+            self.assertEqual([], session.scalars(select(Interval)).all())
+        ws_l.close()
+
+    @gen_test
+    async def test_malformed_frame_content_logging_follows_redaction_setting(self):
+        ws, _ = await self._connect_and_auth(self.admin_id)
+        for redaction, expect_raw in ((True, False), (False, True)):
+            self._app.digi_settings.settings["log_redaction"].set_value(redaction)
+            with self.assertLogs(get_logger(), level="WARNING") as captured:
+                await ws.write_message('{"NOOP": "SENTINEL-FRAME"}')
+                await self._still_responsive(ws, self.admin_id)
+                get_logger().warning("marker")
+            joined = "\n".join(captured.output)
+            self.assertIn("Malformed WS message", joined)
+            self.assertEqual(expect_raw, "SENTINEL-FRAME" in joined)
+        ws.close()
+
+    @gen_test
+    async def test_non_object_and_non_string_op_frames_are_ignored(self):
+        ws, _ = await self._connect_and_auth(self.admin_id)
+        for frame in ("[]", '"abc"', "5", "null", '{"OP": ["x"]}', '{"OP": {}}'):
+            await ws.write_message(frame)
+        await self._still_responsive(ws, self.admin_id)
+        ws.close()
+
+    async def _auth_logs(self, redaction):
+        self._app.digi_settings.settings["log_redaction"].set_value(redaction)
+        ws, _ = await self._connect_and_auth()
+        sentinel = self._app.jwt_service.create_access_token(
+            data={"user_id": self.admin_id}
+        )
+        with self.assertLogs(get_logger(), level="DEBUG") as captured:
+            await ws.write_message(
+                json.dumps({"OP": "AUTHENTICATE", "DATA": {"token": sentinel}})
+            )
+            await self._read_until(ws, op="WS_AUTH_SUCCESS")
+        ws.close()
+        return sentinel, "\n".join(captured.output)
+
+    @gen_test
+    async def test_ws_payload_not_logged_when_redaction_on(self):
+        """With redaction on only the op name and size are logged."""
+        sentinel, joined = await self._auth_logs(True)
+        self.assertIn("AUTHENTICATE", joined)
+        self.assertNotIn(sentinel, joined)
+
+    @gen_test
+    async def test_ws_payload_logged_raw_when_redaction_off(self):
+        sentinel, joined = await self._auth_logs(False)
+        self.assertIn(sentinel, joined)
+
+    @gen_test
+    async def test_update_session_survives_deleted_user(self):
+        ws, uuid = await self._connect_and_auth(self.admin2_id)
+        handler = self._app.get_ws(uuid)
+        with self._app.get_db().sessionmaker() as session:
+            session.get(Session, uuid).user_id = None
+            session.delete(session.get(User, self.admin2_id))
+            session.commit()
+        handler.update_session()  # must not raise
+        ws.close()
+
     @gen_test
     async def test_db_error_in_reconcile_leaves_connection_unauthenticated(self):
         """A failure while settling ownership must not leave the socket half

server/test/digi_server/test_settings.py

@@ -0,0 +1,34 @@
+import json
+
+from test.conftest import DigiScriptTestCase
+
+
+class TestLoggingSettingDefaults(DigiScriptTestCase):
+    def test_log_level_defaults_to_info(self):
+        self.assertEqual("INFO", self._app.digi_settings.settings["log_level"].default)
+
+    def test_log_redaction_enabled_by_default(self):
+        setting = self._app.digi_settings.settings["log_redaction"]
+        self.assertTrue(setting.default)
+        self.assertTrue(setting.get_value())
+
+
+class TestSavedLoggingSettingsAreKept(DigiScriptTestCase):
+    """A value already saved in the settings file wins over a changed default."""
+
+    def get_app(self):
+        with open(self.settings_path, encoding="UTF-8") as file_pointer:
+            saved = json.load(file_pointer)
+        saved["log_redaction"] = False
+        saved["log_level"] = "DEBUG"
+        with open(self.settings_path, "w", encoding="UTF-8") as file_pointer:
+            json.dump(saved, file_pointer)
+        return super().get_app()
+
+    def test_saved_log_redaction_is_not_overridden(self):
+        setting = self._app.digi_settings.settings["log_redaction"]
+        self.assertFalse(setting.get_value())
+
+    def test_saved_log_level_is_not_overridden(self):
+        setting = self._app.digi_settings.settings["log_level"]
+        self.assertEqual("DEBUG", setting.get_value())