Add read-only mode for termui and webui

aiomonitor/monitor.py

@@ -123,6 +123,7 @@ def __init__(
         hook_task_factory: bool = False,
         max_termination_history: int = 1000,
         locals: Optional[Dict[str, Any]] = None,
+        readonly: bool = False,
     ) -> None:
         self._monitored_loop = loop or asyncio.get_running_loop()
         self._host = host
@@ -145,6 +146,8 @@ def __init__(
             },
         )
 
+        self._readonly = readonly
+
         self._closed = False
         self._started = False
         self._termui_tasks = weakref.WeakSet()
@@ -315,6 +318,8 @@ def format_terminated_task_list(
         return tasks
 
     async def cancel_monitored_task(self, task_id: str | int) -> str:
+        if self._readonly:
+            raise PermissionError("Cannot cancel tasks in read-only mode")
         task_id_ = int(task_id)
         task = task_by_id(task_id_, self._monitored_loop)
         if task is not None:
@@ -637,12 +642,15 @@ def start_monitor(
     hook_task_factory: bool = False,
     max_termination_history: Optional[int] = None,
     locals: Optional[Dict[str, Any]] = None,
+    readonly: bool = False,
 ) -> Monitor:
     """
     Factory function, creates instance of :class:`Monitor` and starts
     monitoring thread.
 
-    :param Type[Monitor] monitor: Monitor class to use
+    :param Type[Monitor] monitor: Monitor class to use. This is primarily an
+        internal extension hook; custom monitor classes must keep a constructor
+        signature compatible with :class:`Monitor`, including ``readonly``.
     :param str host: hostname to serve monitor telnet server
     :param int port: monitor port (terminal UI), by default 20101
     :param int webui_port: monitor port (web UI), by default 20102
@@ -651,6 +659,8 @@ def start_monitor(
         to start with instance of monitor.
     :param dict locals: dictionary with variables exposed in python console
         environment
+    :param bool readonly: when True, disables destructive operations like
+        task cancellation, signal sending, and console access.
     """
     m = monitor_cls(
         loop,
@@ -666,6 +676,7 @@ def start_monitor(
             else get_default_args(monitor_cls.__init__)["max_termination_history"]
         ),
         locals=locals,
+        readonly=readonly,
     )
     m.start()
     return m

aiomonitor/termui/commands.py

@@ -35,6 +35,8 @@
 
 log = logging.getLogger(__name__)
 
+_READONLY_MSG = "This command is not available in read-only mode."
+
 __all__ = (
     "interact",
     "monitor_cli",
@@ -89,8 +91,9 @@ async def interact(self: Monitor, connection: TelnetConnection) -> None:
     await asyncio.sleep(0.3)  # wait until telnet negotiation is done
     tasknum = len(asyncio.all_tasks(loop=self._monitored_loop))  # TODO: refactor
     s = "" if tasknum == 1 else "s"
+    mode_indicator = " (read-only)" if self._readonly else ""
     intro = (
-        f"\nAsyncio Monitor: {tasknum} task{s} running\n"
+        f"\nAsyncio Monitor: {tasknum} task{s} running{mode_indicator}\n"
         f"Type help for available commands\n"
     )
     print(intro, file=connection.stdout)
@@ -305,6 +308,10 @@ def do_help(ctx: click.Context) -> None:
 @auto_command_done
 def do_signal(ctx: click.Context, signame: str) -> None:
     """Send a Unix signal"""
+    self: Monitor = ctx.obj
+    if self._readonly:
+        print_fail(_READONLY_MSG)
+        return
     if hasattr(signal, signame):
         os.kill(os.getpid(), getattr(signal, signame))
         print_ok(f"Sent signal to {signame} PID {os.getpid()}")
@@ -331,6 +338,10 @@ def do_stacktrace(ctx: click.Context) -> None:
 def do_cancel(ctx: click.Context, taskid: str) -> None:
     """Cancel an indicated task"""
     self: Monitor = ctx.obj
+    if self._readonly:
+        print_fail(_READONLY_MSG)
+        command_done.get().set()
+        return
 
     @auto_async_command_done
     async def _do_cancel(ctx: click.Context) -> None:
@@ -357,8 +368,13 @@ def do_exit(ctx: click.Context) -> None:
 def do_console(ctx: click.Context) -> None:
     """Switch to async Python REPL"""
     self: Monitor = ctx.obj
+    if self._readonly:
+        print_fail(_READONLY_MSG)
+        command_done.get().set()
+        return
     if not self._console_enabled:
         print_fail("Python console is disabled for this session!")
+        command_done.get().set()
         return
 
     @auto_async_command_done

aiomonitor/webui/app.py

@@ -97,6 +97,7 @@ async def show_list_page(request: web.Request) -> web.Response:
                 {"id": TaskTypes.RUNNING, "title": "Running"},
                 {"id": TaskTypes.TERMINATED, "title": "Terminated"},
             ],
+            readonly=ctx.monitor._readonly,
         )
         return web.Response(body=output, content_type="text/html")
 
@@ -208,6 +209,11 @@ async def get_terminated_task_list(request: web.Request) -> web.Response:
 
 async def cancel_task(request: web.Request) -> web.Response:
     ctx: WebUIContext = request.app[ctx_key]
+    if ctx.monitor._readonly:
+        return web.json_response(
+            status=403,
+            data={"msg": "Cannot cancel tasks in read-only mode"},
+        )
     async with check_params(request, TaskIdParams) as params:
         try:
             coro_repr = await ctx.monitor.cancel_monitored_task(params.task_id)

aiomonitor/webui/templates/index.html

@@ -25,6 +25,21 @@
 </div>
 {% endblock %}
 {% block content %}
+{% if readonly %}
+<style>.task-cancel-btn { display: none !important; }</style>
+<div class="rounded-md bg-yellow-50 p-3 mb-4">
+  <div class="flex">
+    <div class="flex-shrink-0">
+      <svg class="h-5 w-5 text-yellow-400" viewBox="0 0 20 20" fill="currentColor">
+        <path fill-rule="evenodd" d="M8.485 2.495c.673-1.167 2.357-1.167 3.03 0l6.28 10.875c.673 1.167-.17 2.625-1.516 2.625H3.72c-1.347 0-2.189-1.458-1.515-2.625L8.485 2.495zM10 5a.75.75 0 01.75.75v3.5a.75.75 0 01-1.5 0v-3.5A.75.75 0 0110 5zm0 9a1 1 0 100-2 1 1 0 000 2z" clip-rule="evenodd" />
+      </svg>
+    </div>
+    <div class="ml-3">
+      <p class="text-sm font-medium text-yellow-800">Read-only mode: Task cancellation and other destructive operations are disabled.</p>
+    </div>
+  </div>
+</div>
+{% endif %}
 {% if current_list_type == "running" %}
 <div class="flex space-x-2 divide-x divide-gray-200">
   <div class="py-2 px-2">
@@ -115,7 +130,7 @@
     {{^is_root}}
     <button
       type="button"
-      class="notify-result rounded bg-rose-600 px-2 py-1 text-xs font-semibold text-white shadow-sm hover:bg-rose-500 focus-visible:outline focus-visible:outline-2 focus-visible:outline-offset-2 focus-visible:outline-rose-600 disabled:opacity-50"
+      class="task-cancel-btn notify-result rounded bg-rose-600 px-2 py-1 text-xs font-semibold text-white shadow-sm hover:bg-rose-500 focus-visible:outline focus-visible:outline-2 focus-visible:outline-offset-2 focus-visible:outline-rose-600 disabled:opacity-50"
       hx-delete="/api/task"
       hx-params="not persistent,filter"
       hx-vals='{"task_id": "{{ task_id }}"}'

tests/test_monitor.py

@@ -11,6 +11,7 @@
 
 import click
 import pytest
+from aiohttp.test_utils import TestClient, TestServer
 from prompt_toolkit.application import create_app_session
 from prompt_toolkit.input import create_pipe_input
 from prompt_toolkit.output import DummyOutput
@@ -26,10 +27,11 @@
     monitor_cli,
     print_ok,
 )
+from aiomonitor.webui.app import init_webui
 
 
 @contextlib.contextmanager
-def monitor_common():
+def monitor_common(readonly: bool = False):
     def make_baz():
         return "baz"
 
@@ -40,7 +42,7 @@ def make_baz():
     # > fut = asyncio.wrap_future(asyncio.run_coroutine_threadsafe(...))
     # > await fut
     test_loop = asyncio.get_running_loop()
-    mon = Monitor(test_loop, locals=locals_)
+    mon = Monitor(test_loop, locals=locals_, readonly=readonly)
     with mon:
         yield mon
 
@@ -51,6 +53,12 @@ async def monitor(request, event_loop):
         yield monitor_instance
 
 
+@pytest.fixture
+async def readonly_monitor(request, event_loop):
+    with monitor_common(readonly=True) as monitor_instance:
+        yield monitor_instance
+
+
 def get_task_ids(event_loop):
     return [id(t) for t in asyncio.all_tasks(loop=event_loop)]
 
@@ -286,3 +294,81 @@ def do_something(ctx: click.Context, arg: str) -> None:
 
     resp = await invoke_command(monitor, ["something", "someargument"])
     assert "doing something with someargument" in resp
+
+
+async def _create_sleeper_task(monitor: Monitor) -> tuple[asyncio.Task[None], int]:
+    async def sleeper() -> None:
+        await asyncio.sleep(100)
+
+    t = monitor._monitored_loop.create_task(sleeper())
+    t_id = id(t)
+    await asyncio.sleep(0.1)
+    return t, t_id
+
+
+@pytest.mark.asyncio
+async def test_readonly_cancel_termui(readonly_monitor: Monitor) -> None:
+    t, t_id = await _create_sleeper_task(readonly_monitor)
+    resp = await invoke_command(readonly_monitor, ["cancel", str(t_id)])
+    assert "not available in read-only mode" in resp
+    assert not t.done()
+    t.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await t
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "command", [["signal", "SIGUSR1"], ["console"]], ids=["signal", "console"]
+)
+async def test_readonly_simple_termui_commands(
+    readonly_monitor: Monitor, command: list[str]
+) -> None:
+    resp = await invoke_command(readonly_monitor, command)
+    assert "not available in read-only mode" in resp
+
+
+@pytest.mark.asyncio
+async def test_readonly_cancel_monitored_task(readonly_monitor: Monitor) -> None:
+    t, t_id = await _create_sleeper_task(readonly_monitor)
+    with pytest.raises(PermissionError, match="read-only mode"):
+        await readonly_monitor.cancel_monitored_task(t_id)
+    assert not t.done()
+    t.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await t
+
+
+@pytest.mark.asyncio
+async def test_readonly_ps_still_works(readonly_monitor: Monitor) -> None:
+    resp = await invoke_command(readonly_monitor, ["ps"])
+    assert "tasks running" in resp
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("readonly", [True, False])
+async def test_readonly_ctor(readonly: bool) -> None:
+    test_loop = asyncio.get_running_loop()
+    with Monitor(test_loop, readonly=readonly) as m:
+        assert m._readonly is readonly
+        await asyncio.sleep(0.01)
+
+
+@pytest.mark.asyncio
+async def test_readonly_cancel_webui(readonly_monitor: Monitor) -> None:
+    app = await init_webui(readonly_monitor)
+    async with TestClient(TestServer(app)) as client:
+        resp = await client.delete("/api/task", params={"task_id": "123"})
+        assert resp.status == 403
+        data = await resp.json()
+        assert "read-only mode" in data["msg"]
+
+
+@pytest.mark.asyncio
+async def test_webui_cancel_allowed_when_not_readonly(monitor: Monitor) -> None:
+    app = await init_webui(monitor)
+    async with TestClient(TestServer(app)) as client:
+        resp = await client.delete("/api/task", params={"task_id": "123"})
+        # Should not be 403 (will be 500 since cancel_monitored_task
+        # runs on a different loop and the task ID is invalid)
+        assert resp.status != 403