diff --git a/src/central/gui/audit.py b/src/central/gui/audit.py index 520fffc..ada29df 100644 --- a/src/central/gui/audit.py +++ b/src/central/gui/audit.py @@ -14,6 +14,7 @@ STREAM_UPDATE = "stream.update" API_KEY_CREATE = "api_key.create" API_KEY_ROTATE = "api_key.rotate" API_KEY_DELETE = "api_key.delete" +CONSUMER_DELETE = "consumer.delete" SYSTEM_UPDATE = "system.update" MONITORING_AREA_CREATE = "monitoring_area.create" MONITORING_AREA_UPDATE = "monitoring_area.update" diff --git a/src/central/gui/routes.py b/src/central/gui/routes.py index 235f80b..8dff9c5 100644 --- a/src/central/gui/routes.py +++ b/src/central/gui/routes.py @@ -44,6 +44,7 @@ from central.gui.audit import ( AUTH_LOGIN_FAILED, AUTH_LOGOUT, AUTH_PASSWORD_CHANGE, + CONSUMER_DELETE, MONITORING_AREA_CREATE, MONITORING_AREA_DELETE, MONITORING_AREA_UPDATE, @@ -2194,6 +2195,122 @@ async def streams_update( return RedirectResponse(url="/streams", status_code=302) +# ============================================================================= +# Consumers routes +# ============================================================================= + + +@router.get("/consumers", response_class=HTMLResponse) +async def consumers_list(request: Request) -> HTMLResponse: + """List all JetStream consumers across all registered streams.""" + from central.gui.nats import get_js + + templates = _get_templates() + operator = request.state.operator + js = get_js() + + streams_data = [] + for stream_entry in STREAM_REGISTRY: + stream_name = stream_entry.name + consumers = [] + stream_error = None + + if js is not None: + try: + for ci in await js.consumers_info(stream_name): + consumers.append({ + "name": ci.name, + "num_pending": ci.num_pending, + "num_ack_pending": ci.num_ack_pending, + "num_redelivered": ci.num_redelivered, + "num_waiting": ci.num_waiting, + "created": ci.created, + "protected": ci.name.startswith("archive-"), + }) + except Exception as e: + logger.warning( + "consumers_info failed", + extra={"stream": stream_name, "err": type(e).__name__}, + ) + stream_error = f"unavailable: {type(e).__name__}" + else: + stream_error = "NATS unavailable" + + streams_data.append({ + "stream": stream_name, + "consumers": consumers, + "error": stream_error, + }) + + csrf_token = request.state.csrf_token + response = templates.TemplateResponse( + request=request, + name="consumers_list.html", + context={ + "operator": operator, + "csrf_token": csrf_token, + "streams": streams_data, + }, + ) + return response + + +@router.post("/consumers/{stream}/{consumer}/delete", response_class=HTMLResponse) +async def consumers_delete(request: Request, stream: str, consumer: str) -> Response: + """Delete a JetStream consumer.""" + from central.gui.nats import get_js + + operator = request.state.operator + + form = await request.form() + form_csrf = form.get("csrf_token", "") + if not form_csrf or form_csrf != request.state.csrf_token: + raise CsrfValidationError("Invalid CSRF token") + + # Hard guard: never delete archive-* consumers even if path is forged + if consumer.startswith("archive-"): + return RedirectResponse("/consumers", status_code=302) + + js = get_js() + if js is None: + return RedirectResponse("/consumers", status_code=302) + + # Capture before state from NATS for audit log + try: + before_info = await js.consumer_info(stream, consumer) + before = { + "name": before_info.name, + "stream": stream, + "num_pending": before_info.num_pending, + } + except Exception: + before = {"name": consumer, "stream": stream} + + # Delete the consumer + try: + await js.delete_consumer(stream, consumer) + except Exception: + logger.exception( + "delete_consumer failed", + extra={"stream": stream, "consumer": consumer}, + ) + return RedirectResponse("/consumers", status_code=302) + + # Write audit log + pool = get_pool() + async with pool.acquire() as conn: + await write_audit( + conn, + CONSUMER_DELETE, + operator_id=operator.id, + target=f"{stream}/{consumer}", + before=before, + after=None, + ) + + return RedirectResponse("/consumers", status_code=302) + + # ============================================================================= # Enrichment config route # ============================================================================= diff --git a/src/central/gui/templates/base.html b/src/central/gui/templates/base.html index 5732a68..3f2e990 100644 --- a/src/central/gui/templates/base.html +++ b/src/central/gui/templates/base.html @@ -18,6 +18,7 @@ Events Telemetry Streams + Consumers Enrichment Monitoring Area API Keys diff --git a/src/central/gui/templates/consumers_list.html b/src/central/gui/templates/consumers_list.html new file mode 100644 index 0000000..252c1c4 --- /dev/null +++ b/src/central/gui/templates/consumers_list.html @@ -0,0 +1,70 @@ +{% extends "base.html" %} + +{% block title %}Central — Consumers{% endblock %} + +{% block content %} +

Consumers

+

JetStream consumers across all registered streams. A consumer with high +Pending and zero Waiting has accumulated unacknowledged messages and +has no active subscriber — it is safe to delete if it is not a central-owned consumer.

+ +
+{% for stream in streams %} +
+
{{ stream.stream }}
+ + {% if stream.error %} +

({{ stream.error }})

+ {% elif stream.consumers %} + + + + + + + + + + + + + + {% for c in stream.consumers %} + + + + + + + + + + {% endfor %} + +
NamePendingAck PendingRedeliveredWaitingCreatedAction
{{ c.name }}{{ c.num_pending if c.num_pending is not none else '—' }}{{ c.num_ack_pending if c.num_ack_pending is not none else '—' }}{{ c.num_redelivered if c.num_redelivered is not none else '—' }}{{ c.num_waiting if c.num_waiting is not none else '—' }}{{ c.created.isoformat() if c.created else '—' }} + {% if c.protected %} + central-owned + {% else %} +
+ + +
+ {% endif %} +
+ {% else %} +

(no consumers)

+ {% endif %} +
+{% endfor %} +
+ +

+ Legend: Pending = messages not yet delivered to this consumer; + Ack Pending = delivered but not yet acknowledged; + Waiting = active pull requests from a live subscriber. + A consumer with high Pending and zero Waiting is abandoned — no subscriber + is pulling from it and messages are piling up. + Consumers marked central-owned (archive-*) are managed by central and cannot be deleted here. +

+{% endblock %} diff --git a/tests/test_consumers.py b/tests/test_consumers.py new file mode 100644 index 0000000..743dcb8 --- /dev/null +++ b/tests/test_consumers.py @@ -0,0 +1,379 @@ +"""Tests for consumers admin routes (GET /consumers, POST /consumers/{s}/{c}/delete).""" + +import os +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# Set required env vars before importing central modules +os.environ.setdefault("CENTRAL_DB_DSN", "postgresql://test:test@localhost/test") +os.environ.setdefault("CENTRAL_CSRF_SECRET", "testsecret12345678901234567890ab") +os.environ.setdefault("CENTRAL_NATS_URL", "nats://localhost:4222") + + +def _make_consumer_info(name: str, num_pending: int = 0, num_ack_pending: int = 0, + num_redelivered: int = 0, num_waiting: int = 0): + ci = MagicMock() + ci.name = name + ci.num_pending = num_pending + ci.num_ack_pending = num_ack_pending + ci.num_redelivered = num_redelivered + ci.num_waiting = num_waiting + ci.created = datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc) + return ci + + +def _make_js_with_consumers(consumers_by_stream: dict): + """Build a mock JetStreamContext whose consumers_info is a coroutine returning a list.""" + mock_js = MagicMock() + mock_js.consumers_info = AsyncMock( + side_effect=lambda stream, **kw: consumers_by_stream.get(stream, []) + ) + mock_js.consumer_info = AsyncMock() + mock_js.delete_consumer = AsyncMock() + return mock_js + + +class TestConsumersListNatsUnavailable: + """GET /consumers when NATS is down shows per-stream error.""" + + @pytest.mark.asyncio + async def test_nats_unavailable_shows_error_per_stream(self): + from central.gui.routes import consumers_list + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1, username="testop") + mock_request.state.csrf_token = "test_csrf" + + mock_templates = MagicMock() + mock_templates.TemplateResponse.return_value = MagicMock() + + with patch("central.gui.routes._get_templates", return_value=mock_templates): + with patch("central.gui.nats.get_js", return_value=None): + await consumers_list(mock_request) + + call_args = mock_templates.TemplateResponse.call_args + context = call_args.kwargs.get("context", call_args[1].get("context")) + streams = context["streams"] + # All streams should show the NATS unavailable error + assert all(s["error"] == "NATS unavailable" for s in streams) + # And no consumers listed + assert all(s["consumers"] == [] for s in streams) + + +class TestConsumersListWithConsumers: + """GET /consumers with live NATS returns consumers per stream.""" + + @pytest.mark.asyncio + async def test_consumers_listed_with_protected_flag(self): + from central.gui.routes import consumers_list + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1, username="testop") + mock_request.state.csrf_token = "test_csrf" + + mock_templates = MagicMock() + mock_templates.TemplateResponse.return_value = MagicMock() + + consumers_by_stream = { + "CENTRAL_WX": [ + _make_consumer_info("archive-CENTRAL_WX", num_pending=5, num_waiting=1), + _make_consumer_info("meshai-wx", num_pending=1000, num_waiting=0), + ], + } + mock_js = _make_js_with_consumers(consumers_by_stream) + + with patch("central.gui.routes._get_templates", return_value=mock_templates): + with patch("central.gui.nats.get_js", return_value=mock_js): + await consumers_list(mock_request) + + call_args = mock_templates.TemplateResponse.call_args + context = call_args.kwargs.get("context", call_args[1].get("context")) + streams = context["streams"] + + wx = next(s for s in streams if s["stream"] == "CENTRAL_WX") + assert wx["error"] is None + assert len(wx["consumers"]) == 2 + + archive_c = next(c for c in wx["consumers"] if c["name"] == "archive-CENTRAL_WX") + assert archive_c["protected"] is True + assert archive_c["num_pending"] == 5 + + meshai_c = next(c for c in wx["consumers"] if c["name"] == "meshai-wx") + assert meshai_c["protected"] is False + assert meshai_c["num_pending"] == 1000 + assert meshai_c["num_waiting"] == 0 + + # Regression guard: consumer names must appear in the template context so + # they are rendered into the HTML body (guards against the coroutine/iterator + # bug where consumers_info was consumed as an async-iterable instead of awaited). + consumer_names_in_context = {c["name"] for c in wx["consumers"]} + assert "archive-CENTRAL_WX" in consumer_names_in_context + assert "meshai-wx" in consumer_names_in_context + + @pytest.mark.asyncio + async def test_stream_with_no_consumers_shows_empty(self): + from central.gui.routes import consumers_list + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1, username="testop") + mock_request.state.csrf_token = "test_csrf" + + mock_templates = MagicMock() + mock_templates.TemplateResponse.return_value = MagicMock() + + mock_js = _make_js_with_consumers({}) # No consumers on any stream + + with patch("central.gui.routes._get_templates", return_value=mock_templates): + with patch("central.gui.nats.get_js", return_value=mock_js): + await consumers_list(mock_request) + + call_args = mock_templates.TemplateResponse.call_args + context = call_args.kwargs.get("context", call_args[1].get("context")) + streams = context["streams"] + assert all(s["consumers"] == [] for s in streams) + assert all(s["error"] is None for s in streams) + + @pytest.mark.asyncio + async def test_one_stream_error_does_not_break_page(self): + from central.gui.routes import consumers_list + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1, username="testop") + mock_request.state.csrf_token = "test_csrf" + + mock_templates = MagicMock() + mock_templates.TemplateResponse.return_value = MagicMock() + + mock_js = MagicMock() + + async def consumers_info_raising(stream_name): + if stream_name == "CENTRAL_FIRE": + raise RuntimeError("stream not found") + # other streams: empty list (coroutine returning a list, not an async generator) + return [] + + mock_js.consumers_info = consumers_info_raising + + with patch("central.gui.routes._get_templates", return_value=mock_templates): + with patch("central.gui.nats.get_js", return_value=mock_js): + await consumers_list(mock_request) + + call_args = mock_templates.TemplateResponse.call_args + context = call_args.kwargs.get("context", call_args[1].get("context")) + streams = context["streams"] + + fire = next(s for s in streams if s["stream"] == "CENTRAL_FIRE") + assert "unavailable" in fire["error"] + assert fire["consumers"] == [] + + +class TestConsumersListHtmlRender: + """Render consumers_list.html through the real Jinja2 environment. + + Stronger than the context-dict checks above: these prove the values + actually reach the rendered HTML body. Guards two regressions: + - the consumer NAME must appear in the rendered HTML (proves the + ``await js.consumers_info(...)`` list reaches the template, not the + coroutine/async-iterator bug) + - Optional[int] count fields that are None must not render the literal + string ``None`` (they are guarded to an em dash). + """ + + PROTECTED_LABEL = 'central-owned' + + def _render(self, streams): + from central.gui import templates as templates_mod + template = templates_mod.env.get_template("consumers_list.html") + return template.render( + operator=MagicMock(username="testop"), + csrf_token="test_csrf", + streams=streams, + ) + + def test_consumer_name_appears_in_html(self): + streams = [ + { + "stream": "CENTRAL_WX", + "error": None, + "consumers": [ + { + "name": "meshai-wx", + "num_pending": 1000, + "num_ack_pending": 0, + "num_redelivered": 0, + "num_waiting": 0, + "created": datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc), + "protected": False, + }, + ], + }, + ] + html = self._render(streams) + assert "meshai-wx" in html + # Non-protected consumer renders a delete form + assert "/consumers/CENTRAL_WX/meshai-wx/delete" in html + # ...and not the central-owned label span (which only the legend prose + # mentions, so we match the exact span markup, not the bare phrase) + assert self.PROTECTED_LABEL not in html + + def test_protected_consumer_renders_label_not_button(self): + streams = [ + { + "stream": "CENTRAL_WX", + "error": None, + "consumers": [ + { + "name": "archive-CENTRAL_WX", + "num_pending": 5, + "num_ack_pending": 0, + "num_redelivered": 0, + "num_waiting": 1, + "created": datetime(2026, 5, 17, 12, 0, 0, tzinfo=timezone.utc), + "protected": True, + }, + ], + }, + ] + html = self._render(streams) + assert "archive-CENTRAL_WX" in html + assert self.PROTECTED_LABEL in html + # No delete form for the protected consumer + assert "/consumers/CENTRAL_WX/archive-CENTRAL_WX/delete" not in html + + def test_none_counts_render_dash_not_literal_none(self): + streams = [ + { + "stream": "CENTRAL_WX", + "error": None, + "consumers": [ + { + "name": "meshai-wx", + "num_pending": None, + "num_ack_pending": None, + "num_redelivered": None, + "num_waiting": None, + "created": None, + "protected": False, + }, + ], + }, + ] + html = self._render(streams) + assert "meshai-wx" in html + # The literal "None" must never leak into a rendered table cell + assert ">None<" not in html + # The guarded fallback em dash is rendered instead + assert "—" in html + + +class TestConsumersDeleteArchiveGuard: + """POST /consumers/{stream}/archive-*/delete must be refused.""" + + @pytest.mark.asyncio + async def test_archive_consumer_refused_redirects(self): + from central.gui.routes import consumers_delete + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1) + mock_request.state.csrf_token = "tok" + form_data = MagicMock() + form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d) + mock_request.form = AsyncMock(return_value=form_data) + + with patch("central.gui.nats.get_js", return_value=MagicMock()): + result = await consumers_delete(mock_request, "CENTRAL_WX", "archive-CENTRAL_WX") + + assert result.status_code == 302 + assert result.headers["location"] == "/consumers" + + +class TestConsumersDeleteSuccess: + """POST /consumers/{stream}/{consumer}/delete happy path.""" + + @pytest.mark.asyncio + async def test_delete_non_protected_consumer_audits_and_redirects(self): + from central.gui.routes import consumers_delete + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1) + mock_request.state.csrf_token = "tok" + form_data = MagicMock() + form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d) + mock_request.form = AsyncMock(return_value=form_data) + + before_ci = _make_consumer_info("meshai-wx", num_pending=500) + mock_js = MagicMock() + mock_js.consumer_info = AsyncMock(return_value=before_ci) + mock_js.delete_consumer = AsyncMock() + + mock_conn = AsyncMock() + mock_pool = MagicMock() + mock_pool.acquire.return_value.__aenter__ = AsyncMock(return_value=mock_conn) + mock_pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None) + + captured_audit = {} + + async def capture_audit(conn, action, operator_id=None, target=None, before=None, after=None): + captured_audit["action"] = action + captured_audit["operator_id"] = operator_id + captured_audit["target"] = target + captured_audit["before"] = before + captured_audit["after"] = after + + with patch("central.gui.nats.get_js", return_value=mock_js): + with patch("central.gui.routes.get_pool", return_value=mock_pool): + with patch("central.gui.routes.write_audit", side_effect=capture_audit): + result = await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx") + + assert result.status_code == 302 + assert result.headers["location"] == "/consumers" + + mock_js.delete_consumer.assert_awaited_once_with("CENTRAL_WX", "meshai-wx") + + assert captured_audit["action"] == "consumer.delete" + assert captured_audit["operator_id"] == 1 + assert captured_audit["target"] == "CENTRAL_WX/meshai-wx" + assert captured_audit["before"]["name"] == "meshai-wx" + assert captured_audit["after"] is None + + +class TestConsumersDeleteCsrfGuard: + """POST /consumers/{stream}/{consumer}/delete CSRF mismatch raises.""" + + @pytest.mark.asyncio + async def test_csrf_mismatch_raises(self): + from central.gui.routes import consumers_delete + from central.gui.auth import CsrfValidationError + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1) + mock_request.state.csrf_token = "real_token" + form_data = MagicMock() + form_data.get.side_effect = lambda k, d="": {"csrf_token": "wrong_token"}.get(k, d) + mock_request.form = AsyncMock(return_value=form_data) + + with pytest.raises(CsrfValidationError): + await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx") + + +class TestConsumersDeleteNatsUnavailable: + """POST /consumers/{stream}/{consumer}/delete when NATS is down redirects.""" + + @pytest.mark.asyncio + async def test_nats_unavailable_redirects(self): + from central.gui.routes import consumers_delete + + mock_request = MagicMock() + mock_request.state.operator = MagicMock(id=1) + mock_request.state.csrf_token = "tok" + form_data = MagicMock() + form_data.get.side_effect = lambda k, d="": {"csrf_token": "tok"}.get(k, d) + mock_request.form = AsyncMock(return_value=form_data) + + with patch("central.gui.nats.get_js", return_value=None): + result = await consumers_delete(mock_request, "CENTRAL_WX", "meshai-wx") + + assert result.status_code == 302 + assert result.headers["location"] == "/consumers"