From 72d5fa68a3c75868a0b24fb7b9eafe8ef2821a2a Mon Sep 17 00:00:00 2001 From: Clara Vanacker Date: Tue, 30 Jun 2026 21:48:58 +0200 Subject: [PATCH] feat(escalation): auto-remediation action steps (webhook runbooks) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An escalation step can now carry an `action_command`, turning it into an auto-remediation action: when it fires it POSTs a structured remediation intent (the command + problem context, `event: "remediation"`) to its webhook channel for an external runbook to act on. The server never executes commands itself. - EscalationStep.action_command (nullable); EscalationAlertEvent carries it and its payload switches to a remediation shape with the command. - Remediation steps must target a WEBHOOK channel — validated on create (REST + web, 422) and defensively skipped at fire time (logged) if mis-targeted. The once-only step mechanism still applies. - Web form gains an optional per-step "action command" field; the policy list marks remediation steps. Tests: remediation event + payload, non-webhook skip at fire time (step still advances), and the API rejection. Full suite 208 passed; ruff + mypy(strict) green; reversible migration. --- app/api/routes/escalations.py | 11 +++ app/api/routes/web/escalation.py | 17 +++- app/core/models/escalation.py | 5 + app/core/schemas/escalation.py | 3 + app/core/services/escalation_service.py | 54 ++++++++--- app/tasks/notifications/events.py | 13 ++- docs/roadmap.md | 7 +- .../e2a6636ea1da_remediation_step_command.py | 31 ++++++ templates/escalation/list.html | 8 +- tests/test_escalation.py | 94 +++++++++++++++++++ 10 files changed, 219 insertions(+), 24 deletions(-) create mode 100644 migrations/versions/e2a6636ea1da_remediation_step_command.py diff --git a/app/api/routes/escalations.py b/app/api/routes/escalations.py index 42a2eba..8a857a9 100644 --- a/app/api/routes/escalations.py +++ b/app/api/routes/escalations.py @@ -3,6 +3,7 @@ from fastapi import APIRouter, HTTPException, Response, status from app.api.deps import CurrentUser, DBSession +from app.core.models.notification_channel import ChannelType from app.core.schemas.escalation import EscalationPolicyCreate, EscalationPolicyRead from app.core.services.escalation_service import EscalationService @@ -20,6 +21,16 @@ async def _validate_channels( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"unknown or unowned channel(s): {', '.join(str(m) for m in missing)}", ) + # Auto-remediation steps must target a webhook (a machine endpoint). + remediation_channels = {s.channel_id for s in payload.steps if s.action_command} + if remediation_channels: + types = await service.channel_types(owner_id, remediation_channels) + non_webhook = [cid for cid in remediation_channels if types.get(cid) != ChannelType.WEBHOOK] + if non_webhook: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail="auto-remediation steps must target a webhook channel", + ) @router.get("", response_model=list[EscalationPolicyRead], summary="List escalation policies") diff --git a/app/api/routes/web/escalation.py b/app/api/routes/web/escalation.py index a6fdac9..3758d50 100644 --- a/app/api/routes/web/escalation.py +++ b/app/api/routes/web/escalation.py @@ -1,12 +1,14 @@ from __future__ import annotations import uuid +from itertools import zip_longest from fastapi import APIRouter, Form, HTTPException, Request, status from fastapi.responses import HTMLResponse, RedirectResponse from app.api.deps.db import DBSession from app.api.routes.web._shared import login_redirect, resolve_current_user, templates +from app.core.models.notification_channel import ChannelType from app.core.models.user import User from app.core.schemas.escalation import EscalationPolicyCreate, EscalationStepCreate from app.core.services.escalation_service import EscalationService @@ -57,9 +59,12 @@ async def create_escalation_form( form = await request.form() delay_minutes = [str(v) for v in form.getlist("delay_minutes")] channel_id = [str(v) for v in form.getlist("channel_id")] + action_command = [str(v) for v in form.getlist("action_command")] steps: list[EscalationStepCreate] = [] order = 0 - for delay, channel in zip(delay_minutes, channel_id, strict=False): + for delay, channel, command in zip_longest( + delay_minutes, channel_id, action_command, fillvalue="" + ): if not channel: continue order += 1 @@ -69,6 +74,7 @@ async def create_escalation_form( step_order=order, delay_minutes=int(delay or 0), channel_id=uuid.UUID(channel), + action_command=command.strip() or None, ) ) except (ValueError, TypeError): @@ -84,9 +90,14 @@ async def _error(message: str) -> HTMLResponse: if not steps: return await _error("Add at least one step with a channel.") - owned = await service.channels_owned_by(user.id, {s.channel_id for s in steps}) - if owned != {s.channel_id for s in steps}: + wanted = {s.channel_id for s in steps} + if await service.channels_owned_by(user.id, wanted) != wanted: return await _error("One or more selected channels are not yours.") + remediation_channels = {s.channel_id for s in steps if s.action_command} + if remediation_channels: + types = await service.channel_types(user.id, remediation_channels) + if any(types.get(cid) != ChannelType.WEBHOOK for cid in remediation_channels): + return await _error("Auto-remediation steps must target a webhook channel.") try: payload = EscalationPolicyCreate(name=name, is_enabled=(is_enabled == "on"), steps=steps) except (ValueError, TypeError) as exc: diff --git a/app/core/models/escalation.py b/app/core/models/escalation.py index 53bfc29..1ae85cc 100644 --- a/app/core/models/escalation.py +++ b/app/core/models/escalation.py @@ -58,6 +58,11 @@ class EscalationStep(UUIDPrimaryKey, Base): ForeignKey("notification_channels.id", ondelete="CASCADE"), nullable=False, ) + # When set, this is an auto-remediation step: instead of a plain notification it + # POSTs a structured remediation intent (this command + the problem context) to + # its webhook channel, for an external runbook to act on. The server never runs + # commands itself. + action_command: Mapped[str | None] = mapped_column(String(500), nullable=True) policy: Mapped[EscalationPolicy] = relationship(back_populates="steps") channel: Mapped[NotificationChannel] = relationship(lazy="noload") diff --git a/app/core/schemas/escalation.py b/app/core/schemas/escalation.py index ab019b9..ae54868 100644 --- a/app/core/schemas/escalation.py +++ b/app/core/schemas/escalation.py @@ -10,6 +10,8 @@ class EscalationStepCreate(BaseModel): step_order: int = Field(ge=1, le=100) delay_minutes: int = Field(ge=0, le=10080) # up to a week channel_id: uuid.UUID + # When set, an auto-remediation step: POSTs this command (+ context) to a webhook. + action_command: str | None = Field(default=None, max_length=500) class EscalationStepRead(BaseModel): @@ -19,6 +21,7 @@ class EscalationStepRead(BaseModel): step_order: int delay_minutes: int channel_id: uuid.UUID + action_command: str | None class EscalationPolicyCreate(BaseModel): diff --git a/app/core/services/escalation_service.py b/app/core/services/escalation_service.py index b7cf631..959d93b 100644 --- a/app/core/services/escalation_service.py +++ b/app/core/services/escalation_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import uuid from collections.abc import Sequence from datetime import datetime @@ -9,11 +10,13 @@ from sqlalchemy.orm import selectinload from app.core.models.escalation import EscalationPolicy, EscalationStep -from app.core.models.notification_channel import NotificationChannel +from app.core.models.notification_channel import ChannelType, NotificationChannel from app.core.models.problem_event import ProblemEvent from app.core.schemas.escalation import EscalationPolicyCreate from app.tasks.notifications.events import EscalationAlertEvent +logger = logging.getLogger(__name__) + class EscalationService: def __init__(self, session: AsyncSession) -> None: @@ -44,6 +47,7 @@ async def create(self, owner_id: uuid.UUID, data: EscalationPolicyCreate) -> Esc step_order=step.step_order, delay_minutes=step.delay_minutes, channel_id=step.channel_id, + action_command=step.action_command, ) ) self._session.add(policy) @@ -62,6 +66,7 @@ async def replace( step_order=step.step_order, delay_minutes=step.delay_minutes, channel_id=step.channel_id, + action_command=step.action_command, ) ) await self._session.commit() @@ -88,6 +93,18 @@ async def channels_owned_by( ) return set((await self._session.execute(stmt)).scalars().all()) + async def channel_types( + self, owner_id: uuid.UUID, channel_ids: set[uuid.UUID] + ) -> dict[uuid.UUID, ChannelType]: + """Map the owner's channels (among `channel_ids`) to their type.""" + if not channel_ids: + return {} + stmt = select(NotificationChannel.id, NotificationChannel.type).where( + NotificationChannel.owner_id == owner_id, + NotificationChannel.id.in_(channel_ids), + ) + return {cid: ctype for cid, ctype in (await self._session.execute(stmt))} + async def _enabled_policy(self, owner_id: uuid.UUID) -> EscalationPolicy | None: stmt = ( select(EscalationPolicy) @@ -132,21 +149,28 @@ async def due_escalations( break # later steps have larger delays channel = await self._session.get(NotificationChannel, step.channel_id) if channel is not None and channel.is_enabled: - deliveries.append( - ( - EscalationAlertEvent( - problem_id=problem.id, - subject=problem.subject, - trigger_name=problem.trigger_name, - severity=problem.severity, - step_order=step.step_order, - value=problem.value, - started_at=problem.started_at, - timestamp=now, - ), - channel, + # Auto-remediation must reach a machine endpoint, never an inbox. + if step.action_command is not None and channel.type != ChannelType.WEBHOOK: + logger.warning( + "remediation step %s targets a non-webhook channel; skipping", step.id + ) + else: + deliveries.append( + ( + EscalationAlertEvent( + problem_id=problem.id, + subject=problem.subject, + trigger_name=problem.trigger_name, + severity=problem.severity, + step_order=step.step_order, + value=problem.value, + started_at=problem.started_at, + timestamp=now, + action_command=step.action_command, + ), + channel, + ) ) - ) problem.escalated_step = step.step_order await self._session.commit() return deliveries diff --git a/app/tasks/notifications/events.py b/app/tasks/notifications/events.py index ad1fa9d..a72a787 100644 --- a/app/tasks/notifications/events.py +++ b/app/tasks/notifications/events.py @@ -147,14 +147,20 @@ class EscalationAlertEvent: value: float | None started_at: datetime timestamp: datetime + # When set, this step is an auto-remediation action (not a plain notification). + action_command: str | None = None @property def is_recovery(self) -> bool: return False + @property + def is_remediation(self) -> bool: + return self.action_command is not None + def payload(self) -> dict[str, Any]: - return { - "event": "escalation", + body: dict[str, Any] = { + "event": "remediation" if self.is_remediation else "escalation", "severity": self.severity.value, "step": self.step_order, "subject": self.subject, @@ -163,3 +169,6 @@ def payload(self) -> dict[str, Any]: "started_at": self.started_at.isoformat(), "timestamp": self.timestamp.isoformat(), } + if self.action_command is not None: + body["action"] = {"command": self.action_command} + return body diff --git a/docs/roadmap.md b/docs/roadmap.md index c71b6c9..0fb2a58 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -114,8 +114,11 @@ GhostMonitor's reason to exist over a plain Zabbix clone is privacy (ghost-suite - ✅ **Hosts overview**: the host list shows each host's health (item count, ongoing problem count and worst severity), problem hosts sorted first — aggregated in two grouped queries (no N+1). -- *(next)* long-range charts backed by trends; auto-remediation hooks as an - escalation action. +- ✅ **Auto-remediation actions**: an escalation step can carry an `action_command`, + making it POST a structured remediation intent (command + problem context) to a + **webhook** channel for an external runbook — the server never runs commands itself + (webhook-only, validated on create and at fire time). +- *(next)* long-range charts backed by trends. ## Phase 5 — Discovery & scale diff --git a/migrations/versions/e2a6636ea1da_remediation_step_command.py b/migrations/versions/e2a6636ea1da_remediation_step_command.py new file mode 100644 index 0000000..63cbb51 --- /dev/null +++ b/migrations/versions/e2a6636ea1da_remediation_step_command.py @@ -0,0 +1,31 @@ +"""remediation step command + +Revision ID: e2a6636ea1da +Revises: 0414a77d051d +Create Date: 2026-06-30 21:44:14.940544 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "e2a6636ea1da" +down_revision: str | Sequence[str] | None = "0414a77d051d" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.add_column( + "escalation_steps", sa.Column("action_command", sa.String(length=500), nullable=True) + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_column("escalation_steps", "action_command") + # ### end Alembic commands ### diff --git a/templates/escalation/list.html b/templates/escalation/list.html index bec6a95..85be036 100644 --- a/templates/escalation/list.html +++ b/templates/escalation/list.html @@ -28,7 +28,7 @@

Policies {{ policies|length }}

{{ p.name }} {{ 'yes' if p.is_enabled else 'no' }} - {% for s in p.steps %}{{ s.delay_minutes }}m → {{ channel_names.get(s.channel_id, '?') }}{% if not loop.last %} · {% endif %}{% endfor %} + {% for s in p.steps %}{{ s.delay_minutes }}m → {{ channel_names.get(s.channel_id, '?') }}{% if s.action_command %} ⚙{% endif %}{% if not loop.last %} · {% endif %}{% endfor %}
New policy Enabled -

Each step notifies a channel after the delay (in minutes) from when the problem opened. Leave a channel empty to skip the row.

+

Each step notifies a channel after the delay (in minutes) from when the problem opened. Leave a channel empty to skip the row. Set an action command to make a step auto-remediation: it POSTs the command + problem context to a webhook channel for an external runbook (the server never runs commands itself).

{% for i in step_rows %}
+
{% endfor %}
diff --git a/tests/test_escalation.py b/tests/test_escalation.py index 198a216..9f2076a 100644 --- a/tests/test_escalation.py +++ b/tests/test_escalation.py @@ -209,3 +209,97 @@ async def test_escalation_web_requires_a_step( resp = await web_client.post("/escalation/new", content=body, headers=_FORM_HEADERS) assert resp.status_code == 422 assert "at least one step" in resp.text.lower() + + +async def test_remediation_step_emits_remediation_event(session: Any, user: Any) -> None: + hook = await _channel(session, user.id, "automation") # webhook + trig = await _trigger(session, user.id) + await EscalationService(session).create( + user.id, + EscalationPolicyCreate( + name="auto", + steps=[ + EscalationStepCreate( + step_order=1, + delay_minutes=0, + channel_id=hook.id, + action_command="restart nginx", + ) + ], + ), + ) + await _problem(session, user.id, trig.id, NOW - timedelta(minutes=1)) + await session.commit() + + deliveries = await EscalationService(session).due_escalations(NOW) + assert len(deliveries) == 1 + event, _channel_obj = deliveries[0] + assert event.is_remediation + assert event.action_command == "restart nginx" + payload = event.payload() + assert payload["event"] == "remediation" + assert payload["action"]["command"] == "restart nginx" + + +async def test_remediation_step_skipped_for_non_webhook_channel(session: Any, user: Any) -> None: + email = NotificationChannel( + name="inbox", + type=ChannelType.EMAIL, + config={"to": "ops@x.io"}, + owner_id=user.id, + min_severity=Severity.INFO, + ) + session.add(email) + await session.flush() + trig = await _trigger(session, user.id) + await EscalationService(session).create( + user.id, + EscalationPolicyCreate( + name="bad-auto", + steps=[ + EscalationStepCreate( + step_order=1, delay_minutes=0, channel_id=email.id, action_command="rm -rf" + ) + ], + ), + ) + pe = await _problem(session, user.id, trig.id, NOW - timedelta(minutes=1)) + await session.commit() + + # The engine refuses to send a remediation to an inbox, but still advances the + # step so it is not retried every minute. + assert await EscalationService(session).due_escalations(NOW) == [] + await session.refresh(pe) + assert pe.escalated_step == 1 + + +async def test_api_rejects_remediation_on_non_webhook( + client: httpx.AsyncClient, auth_headers: dict[str, str], session: Any, user: Any +) -> None: + email = NotificationChannel( + name="inbox", + type=ChannelType.EMAIL, + config={"to": "ops@x.io"}, + owner_id=user.id, + min_severity=Severity.INFO, + ) + session.add(email) + await session.commit() + + resp = await client.post( + "/api/escalation-policies", + headers=auth_headers, + json={ + "name": "bad", + "steps": [ + { + "step_order": 1, + "delay_minutes": 0, + "channel_id": str(email.id), + "action_command": "restart", + } + ], + }, + ) + assert resp.status_code == 422 + assert "webhook" in resp.text.lower()