feat: add PRTG Network Monitor provider

keep/providers/prtg_provider/alerts_mock.py

@@ -0,0 +1,19 @@
+{
+    "host": "192.168.1.100",
+    "name": "Ping Sensor",
+    "sensor": "Ping",
+    "message": "Sensor is down",
+    "status": "Down",
+    "lastvalue": "0 ms",
+    "device": "Server-01",
+    "group": "Production",
+    "probe": "Probe1",
+    "link": "https://prtg.example.com/sensor.htm?id=1234",
+    "id": "1234",
+    "sensorid": "1234",
+    "datetime": "2026-05-22 10:30:00",
+    "down": "5 minutes",
+    "sensor_type": "Ping",
+    "tags": "production,critical",
+    "priority": "5"
+}

keep/providers/prtg_provider/prtg_provider.py

@@ -0,0 +1,215 @@
+"""
+PRTG (Paessler Router Traffic Grapher) is a network monitoring tool that provides
+real-time monitoring of servers, applications, and network devices.
+"""
+
+import dataclasses
+
+import pydantic
+
+from keep.api.models.alert import AlertDto, AlertSeverity, AlertStatus
+from keep.contextmanager.contextmanager import ContextManager
+from keep.providers.base.base_provider import BaseProvider
+from keep.providers.models.provider_config import ProviderConfig, ProviderScope
+
+
+@pydantic.dataclasses.dataclass
+class PrtgProviderAuthConfig:
+    """
+    PRTG authentication configuration.
+    PRTG uses HTTP Action notifications to push alerts to external systems.
+    No authentication is required on the receiving end — PRTG sends the data
+    via HTTP POST to the configured webhook URL.
+    """
+
+    pass
+
+
+class PrtgProvider(BaseProvider):
+    """Receive alerts from PRTG Network Monitor via HTTP Action notifications."""
+
+    PROVIDER_DISPLAY_NAME = "PRTG"
+    PROVIDER_CATEGORY = ["Monitoring"]
+    PROVIDER_TAGS = ["alert"]
+    FINGERPRINT_FIELDS = ["id"]
+    PROVIDER_SCOPES = [
+        ProviderScope(
+            name="connected",
+            description="Provider is connected and can receive alerts",
+            mandatory=False,
+            mandatory_for_webhook=False,
+            alias="Connected",
+        ),
+    ]
+
+    webhook_description = "PRTG sends alert notifications via HTTP Action (HTTP POST)."
+    webhook_template = ""
+    webhook_markdown = """
+To send alerts from PRTG to Keep, configure an HTTP Action notification in PRTG:
+
+1. In PRTG, go to **Setup** → **Account Settings** → **Notification Templates**.
+2. Click **Add Notification Template**.
+3. Set a name, e.g. "Keep".
+4. Enable **Execute HTTP Action**.
+5. Set **URL** to: `{keep_webhook_api_url}`
+6. Set **HTTP Method** to: `POST`
+7. Set **Payload** to the following JSON:
+```json
+{{
+  "host": "%%host",
+  "name": "%%name",
+  "sensor": "%%sensor",
+  "message": "%%message",
+  "status": "%%status",
+  "lastvalue": "%%lastvalue",
+  "device": "%%device",
+  "group": "%%group",
+  "probe": "%%probe",
+  "link": "%%link",
+  "id": "%%id",
+  "sensorid": "%%sensorid",
+  "datetime": "%%datetime",
+  "down": "%%down",
+  "sensor_type": "%%sensor_type",
+  "tags": "%%tags",
+  "priority": "%%priority"
+}}
+```
+8. Click **Save**.
+9. Assign this notification template to your sensors/devices via **Notification Triggers**.
+"""
+
+    SEVERITIES_MAP = {
+        "1": AlertSeverity.INFO,       # Info
+        "2": AlertSeverity.WARNING,    # Warning
+        "3": AlertSeverity.WARNING,    # Unusual
+        "4": AlertSeverity.HIGH,       # High
+        "5": AlertSeverity.CRITICAL,   # Down / Critical
+    }
+
+    STATUS_MAP = {
+        "Up": AlertStatus.RESOLVED,
+        "Warning": AlertStatus.FIRING,
+        "Down": AlertStatus.FIRING,
+        "Paused": AlertStatus.SUPPRESSED,
+        "Unknown": AlertStatus.FIRING,
+    }
+
+    def __init__(
+        self, context_manager: ContextManager, provider_id: str, config: ProviderConfig
+    ):
+        super().__init__(context_manager, provider_id, config)
+
+    def dispose(self):
+        """
+        Dispose the provider.
+        """
+        pass
+
+    def validate_config(self):
+        """
+        Validates required configuration for PRTG provider.
+        PRTG is a webhook-only provider with no required auth fields.
+        """
+        if self.config.authentication:
+            self.authentication_config = PrtgProviderAuthConfig(
+                **self.config.authentication
+            )
+
+    def validate_scopes(self) -> dict[str, bool | str]:
+        """
+        Validate scopes for PRTG provider.
+        PRTG is a webhook-only provider, so we just confirm connectivity.
+        """
+        return {"connected": True}
+
+    @staticmethod
+    def _format_alert(
+        event: dict, provider_instance: "BaseProvider" = None
+    ) -> AlertDto:
+        """
+        Format a PRTG alert event into an AlertDto.
+
+        PRTG sends alerts via HTTP POST with the following fields (using %% placeholders):
+        - host: The hostname or IP of the monitored device
+        - name: The name of the sensor
+        - sensor: The sensor name
+        - message: The alert message
+        - status: The current status (Up, Down, Warning, Paused, Unknown)
+        - lastvalue: The last measured value
+        - device: The device name
+        - group: The group name
+        - probe: The probe name
+        - link: URL to the sensor in PRTG
+        - id: The sensor ID
+        - sensorid: The sensor ID (alternative)
+        - datetime: The date and time of the alert
+        - down: Duration the sensor has been down
+        - sensor_type: The type of sensor
+        - tags: Tags associated with the sensor
+        - priority: Priority level (1-5)
+        """
+        # Extract key fields
+        sensor_name = event.get("sensor") or event.get("name") or "Unknown Sensor"
+        device_name = event.get("device") or event.get("host") or "Unknown Device"
+        message = event.get("message") or ""
+        status = event.get("status") or "Unknown"
+        link = event.get("link") or ""
+        # PRTG may send unresolved placeholders (e.g. "%%link") when not configured
+        # These are not valid URLs and will cause ValidationError on AlertDto.url
+        if link.startswith("%%"):
+            link = None
+        sensor_id = (
+            event.get("sensorid") if event.get("sensorid") is not None
+            else event.get("id") or ""
+        )
+        priority = event.get("priority") or ""
+
+        # Build alert name: "Device - Sensor"
+        alert_name = f"{device_name} - {sensor_name}"
+
+        # Map severity from PRTG priority (1-5) or status
+        severity = PrtgProvider.SEVERITIES_MAP.get(
+            str(priority),
+            # Fallback: map from status
+            {
+                "Down": AlertSeverity.CRITICAL,
+                "Warning": AlertSeverity.WARNING,
+                "Up": AlertSeverity.INFO,
+                "Paused": AlertSeverity.INFO,
+                "Unknown": AlertSeverity.INFO,
+            }.get(status, AlertSeverity.INFO),
+        )
+
+        # Map status
+        alert_status = PrtgProvider.STATUS_MAP.get(status, AlertStatus.FIRING)
+
+        # Build the AlertDto
+        # AlertDto uses Config with extra = Extra.allow, so additional
+        # PRTG-specific fields are stored as extra attributes.
+        alert = AlertDto(
+            id=sensor_id,
+            name=alert_name,
+            message=message,
+            severity=severity,
+            status=alert_status,
+            url=link,
+            lastReceived=event.get("datetime"),
+            source=["prtg"],
+            # PRTG-specific extra fields (allowed by Extra.allow)
+            sensor=sensor_name,
+            device=device_name,
+            group=event.get("group"),
+            probe=event.get("probe"),
+            lastvalue=event.get("lastvalue"),
+            down=event.get("down"),
+            sensor_type=event.get("sensor_type"),
+            tags=event.get("tags"),
+            priority=priority,
+        )
+
+        return alert
+
+
+if __name__ == "__main__":
+    pass

tests/providers/prtg_provider/test_prtg_provider.py

@@ -0,0 +1,142 @@
+import pytest
+
+from keep.providers.prtg_provider.prtg_provider import PrtgProvider
+
+
+# Sample full PRTG webhook event
+FULL_EVENT = {
+    "host": "192.168.1.100",
+    "name": "Ping Sensor",
+    "sensor": "Ping",
+    "message": "Sensor is down",
+    "status": "Down",
+    "lastvalue": "0 ms",
+    "device": "Server-01",
+    "group": "Production",
+    "probe": "Probe1",
+    "link": "https://prtg.example.com/sensor.htm?id=1234",
+    "id": "1234",
+    "sensorid": "1234",
+    "datetime": "2026-05-22 10:30:00",
+    "down": "5 minutes",
+    "sensor_type": "Ping",
+    "tags": "production,critical",
+    "priority": "5",
+}
+
+
+class TestPrtgFormatAlert:
+    """Tests for PrtgProvider._format_alert."""
+
+    def test_format_alert_full_event(self):
+        alert = PrtgProvider._format_alert(FULL_EVENT)
+        assert alert.id == "1234"
+        assert alert.name == "Server-01 - Ping"
+        assert alert.message == "Sensor is down"
+        assert alert.severity.value == "critical"
+        assert alert.status.value == "firing"
+        assert str(alert.url) == "https://prtg.example.com/sensor.htm?id=1234"
+        assert alert.lastReceived == "2026-05-22 10:30:00"
+        assert alert.source == ["prtg"]
+        assert alert.sensor == "Ping"
+        assert alert.device == "Server-01"
+        assert alert.group == "Production"
+        assert alert.probe == "Probe1"
+        assert alert.lastvalue == "0 ms"
+        assert alert.down == "5 minutes"
+        assert alert.sensor_type == "Ping"
+        assert alert.tags == "production,critical"
+        assert alert.priority == "5"
+
+    def test_format_alert_minimal_event(self):
+        """Only required fields — everything else should have safe defaults."""
+        event = {"sensor": "CPU Load", "device": "Router-01", "status": "Down"}
+        alert = PrtgProvider._format_alert(event)
+        assert alert.name == "Router-01 - CPU Load"
+        assert alert.severity.value == "critical"
+        assert alert.status.value == "firing"
+        assert alert.url is None
+        assert alert.id == ""
+
+    def test_format_alert_status_mapping(self):
+        """All PRTG statuses map correctly."""
+        cases = {
+            "Up": ("resolved", "info"),
+            "Down": ("firing", "critical"),
+            "Warning": ("firing", "warning"),
+            "Paused": ("suppressed", "info"),
+            "Unknown": ("firing", "info"),
+        }
+        for status, (expected_status, expected_severity) in cases.items():
+            event = dict(FULL_EVENT, status=status, priority="")
+            alert = PrtgProvider._format_alert(event)
+            assert alert.status.value == expected_status, f"status={status}"
+            assert alert.severity.value == expected_severity, f"status={status}"
+
+    def test_format_alert_severity_from_priority(self):
+        """Priority 1-5 maps to correct severity levels."""
+        cases = {
+            "1": "info",
+            "2": "warning",
+            "3": "warning",
+            "4": "high",
+            "5": "critical",
+        }
+        for priority, expected in cases.items():
+            event = dict(FULL_EVENT, priority=priority, status="Unknown")
+            alert = PrtgProvider._format_alert(event)
+            assert alert.severity.value == expected, f"priority={priority}"
+
+    def test_format_alert_severity_fallback_to_status(self):
+        """When priority is missing/empty, severity falls back to status mapping."""
+        event = dict(FULL_EVENT, priority="", status="Down")
+        alert = PrtgProvider._format_alert(event)
+        assert alert.severity.value == "critical"
+
+        event2 = dict(FULL_EVENT, priority=None, status="Warning")
+        alert2 = PrtgProvider._format_alert(event2)
+        assert alert2.severity.value == "warning"
+
+    def test_format_alert_unresolved_placeholders(self):
+        """PRTG may send %%placeholder when values aren't configured.
+        These must be sanitized — especially %%link which would cause
+        ValidationError on AlertDto.url (AnyHttpUrl type)."""
+        event = dict(FULL_EVENT, link="%%link", sensor_type="%%sensor_type")
+        alert = PrtgProvider._format_alert(event)
+        assert alert.url is None  # %%link sanitized to None
+        # Other %% fields are just stored as strings (Extra.allow)
+        assert alert.sensor_type == "%%sensor_type"
+
+    def test_format_alert_sensorid_precedence(self):
+        """sensorid takes precedence over id when both present."""
+        event = dict(FULL_EVENT, sensorid="999", id="1234")
+        alert = PrtgProvider._format_alert(event)
+        assert alert.id == "999"
+
+    def test_format_alert_sensorid_zero(self):
+        """sensorid='0' should not fall through to id (0 is falsy but valid)."""
+        event = dict(FULL_EVENT, sensorid="0", id="1234")
+        alert = PrtgProvider._format_alert(event)
+        assert alert.id == "0"
+
+    def test_format_alert_empty_event(self):
+        """Empty dict should not crash — safe defaults for everything."""
+        alert = PrtgProvider._format_alert({})
+        assert alert.name == "Unknown Device - Unknown Sensor"
+        assert alert.status.value == "firing"
+        assert alert.severity.value == "info"
+        assert alert.url is None
+        assert alert.id == ""
+
+    def test_format_alert_numeric_priority(self):
+        """PRTG might send priority as integer instead of string."""
+        event = dict(FULL_EVENT, priority=5)
+        alert = PrtgProvider._format_alert(event)
+        assert alert.severity.value == "critical"
+
+    def test_format_alert_host_as_device_fallback(self):
+        """When device is missing, host is used as device name."""
+        event = {"host": "10.0.0.1", "sensor": "Ping", "status": "Up"}
+        alert = PrtgProvider._format_alert(event)
+        assert "10.0.0.1" in alert.name
+        assert alert.device == "10.0.0.1"