FIX REST API: PagerDuty notification rule integration key format

.werks/20007.md

@@ -0,0 +1,30 @@
+[//]: # (werk v3)
+# REST API: Fix PagerDuty notification rule integration key format
+
+key        | value
+---------- | ---
+date       | 2026-05-19T12:00:00+00:00
+version    | 2.6.0b1
+class      | fix
+edition    | community
+component  | rest-api
+level      | 1
+compatible | yes
+
+PagerDuty notification rules created via the REST API silently produced broken
+on-disk configuration.  When such a rule was triggered the notification
+dispatcher exited with `Unable to retrieve password from passwordstore` and the
+alert was lost.
+
+The REST API was writing the integration key to `notification_parameter.mk`
+as the legacy two-element tuple `("routing_key", key)`.  The notification
+dispatcher expects the modern three-element form
+`("cmk_postprocessed", "explicit_password", (uuid, key))`, which the GUI's
+FormSpec migration produces but which the REST API path bypassed.
+
+The REST API now serialises the integration key in the same form as the GUI.
+Rules created or updated via the API are dispatched correctly.
+
+Rules created in earlier patch levels whose on-disk values are still in the
+legacy form are normalised on read, so they keep working through the API and
+are rewritten in the new format the next time they are saved.

.werks/first_free

@@ -1 +1 @@
-20007
+20008

cmk/gui/rest_api_types/notifications_rule_types.py

@@ -67,7 +67,6 @@
     PushOverPriorityStringType,
     PushOverPriorityType,
     RegexModes,
-    RoutingKeyType,
     ServiceEventType,
     ServiceNowPluginName,
     Signl4PluginName,
@@ -2494,47 +2493,6 @@ def to_mk_file_format(self) -> CheckmkPassword | None:
         return self.checkmk_password
 
 
-@dataclass
-class APIPagerDutyKeyOption:
-    option: Literal["explicit", "store"] | None = None
-    store_id: str = ""
-    key: str = ""
-
-    @classmethod
-    def from_api_request(cls, incoming: APIKey) -> APIPagerDutyKeyOption:
-        if "key" in incoming:
-            return cls(option="explicit", key=incoming["key"])
-        return cls(option="store", store_id=incoming["store_id"])
-
-    def api_response(self) -> APIKey:
-        if self.option is None:
-            return {}
-
-        r: APIKey = {"option": self.option}
-        if self.option == "explicit":
-            r["key"] = self.key
-            return r
-        r["store_id"] = self.store_id
-        return r
-
-    @classmethod
-    def from_mk_file_format(cls, data: RoutingKeyType | None) -> APIPagerDutyKeyOption:
-        if data is None:
-            return cls()
-
-        if "routing_key" in data:
-            return cls(option="explicit", key=data[1])
-        return cls(option="store", key=data[1])
-
-    def to_mk_file_format(self) -> RoutingKeyType | None:
-        if self.option is None:
-            return None
-
-        if self.option == "explicit":
-            return "routing_key", self.key
-        return "store", self.store_id
-
-
 # ----------------------------------------------------------------
 class API_WebhookURL(API_ExplicitOrStore, total=False):
     url: str

cmk/gui/rest_api_types/notifications_types.py

@@ -37,7 +37,6 @@
     APICheckmkPassword_FromPassword,
     APICheckmkPassword_FromSecret,
     APINotifyPlugin,
-    APIPagerDutyKeyOption,
     APIPluginDict,
     BasicOrTokenAuth,
     CheckboxEmailBodyInfo,
@@ -1176,11 +1175,34 @@ def to_mk_file_format(self) -> PluginNameWithParameters:
         )
 
 
+def _normalize_pagerduty_routing_key(data: object) -> CheckmkPassword:
+    """Accept both the legacy 2-tuple and the current CheckmkPassword 3-tuple.
+
+    Rules created via the REST API in 2.4.0 wrote the legacy `("routing_key", key)`
+    or `("store", store_id)` form. The notification dispatcher expects the
+    `("cmk_postprocessed", "explicit_password", (uuid, key))` form. The GUI's
+    FormSpec migrate covers UI-touched rules; this normalizer covers rules read
+    via the REST API that have not yet been rewritten through the UI.
+    """
+    if isinstance(data, tuple):
+        if len(data) == 2 and data[0] == "routing_key":
+            return (
+                "cmk_postprocessed",
+                "explicit_password",
+                (password_store.ad_hoc_password_id(), data[1]),
+            )
+        if len(data) == 2 and data[0] == "store":
+            return ("cmk_postprocessed", "stored_password", (data[1], ""))
+        if len(data) == 3 and data[0] == "cmk_postprocessed":
+            return cast(CheckmkPassword, data)
+    raise ValueError(f"Invalid PagerDuty routing_key format: {data!r}")
+
+
 @dataclass
 class PagerDutyPlugin:
     plugin_name: ClassVar[PagerdutyPluginName] = "pagerduty"
     option: PluginOptions = PluginOptions.CANCEL
-    integration_key: APIPagerDutyKeyOption = field(default_factory=APIPagerDutyKeyOption)
+    integration_key: APICheckmkPassword_FromKey = field(default_factory=APICheckmkPassword_FromKey)
     disable_ssl_cert_verification: CheckboxTrueOrNone = field(default_factory=CheckboxTrueOrNone)
     http_proxy: CheckboxHttpProxy = field(default_factory=CheckboxHttpProxy)
     url_prefix_for_links_to_checkmk: CheckboxURLPrefix = field(default_factory=CheckboxURLPrefix)
@@ -1195,7 +1217,9 @@ def from_mk_file_format(cls, pluginparams: PagerDutyPluginModel | None) -> Pager
 
         return cls(
             option=PluginOptions.WITH_PARAMS,
-            integration_key=APIPagerDutyKeyOption.from_mk_file_format(pluginparams["routing_key"]),
+            integration_key=APICheckmkPassword_FromKey.from_mk_file_format(
+                _normalize_pagerduty_routing_key(pluginparams["routing_key"]),
+            ),
             disable_ssl_cert_verification=CheckboxTrueOrNone.from_mk_file_format(
                 pluginparams.get("ignore_ssl"),
             ),
@@ -1216,7 +1240,7 @@ def from_api_request(cls, incoming: APINotifyPlugin) -> PagerDutyPlugin:
 
         return cls(
             option=PluginOptions.WITH_PARAMS,
-            integration_key=APIPagerDutyKeyOption.from_api_request(params["integration_key"]),
+            integration_key=APICheckmkPassword_FromKey.from_api_request(params["integration_key"]),
             disable_ssl_cert_verification=CheckboxTrueOrNone.from_api_request(
                 params["disable_ssl_cert_verification"]
             ),

cmk/utils/notify_types.py

@@ -592,7 +592,7 @@ class JsmOperationsPluginModel(TypedDict, total=False):
 
 
 class PagerDutyPluginModel(TypedDict):
-    routing_key: tuple[Literal["routing_key", "store"], str]
+    routing_key: CheckmkPassword
     webhook_url: Literal["https://events.pagerduty.com/v2/enqueue"]
     ignore_ssl: NotRequired[Literal[True]]
     proxy_url: NotRequired[ProxyUrl]

tests/unit/cmk/gui/watolib/test_notifications.py

@@ -265,3 +265,93 @@ def test_notification_rule_roundtrip_preserves_explicit_email_addresses_handling
 
     assert mk_format_cloud["contact_users"] == ["user1"]
     assert mk_format_cloud["contact_emails"] == ["cloud@example.com"]
+
+
+def test_pagerduty_from_api_request_produces_cmk_postprocessed_tuple() -> None:
+    """A PagerDuty rule created via the REST API must serialize to the
+    `("cmk_postprocessed", "explicit_password", (uuid, key))` form so the
+    notification dispatcher recognises the `"explicit_password"` marker.
+    """
+    from cmk.gui.rest_api_types.notifications_rule_types import (
+        API_PagerDutyData,
+        APINotifyPlugin,
+    )
+    from cmk.gui.rest_api_types.notifications_types import PagerDutyPlugin
+    from cmk.utils.notify_types import PluginOptions
+
+    incoming: APINotifyPlugin = {
+        "option": PluginOptions.WITH_PARAMS,
+        "plugin_params": API_PagerDutyData(
+            plugin_name="pagerduty",
+            integration_key={"option": "explicit", "key": "abcdef0123456789"},
+            disable_ssl_cert_verification={"state": "disabled"},
+            http_proxy={"state": "disabled"},
+            url_prefix_for_links_to_checkmk={"state": "disabled"},
+        ),
+    }
+
+    plugin = PagerDutyPlugin.from_api_request(incoming)
+    _, params = plugin.to_mk_file_format()
+    assert params is not None
+    routing_key = params["routing_key"]
+    assert routing_key[0] == "cmk_postprocessed"
+    assert routing_key[1] == "explicit_password"
+    assert routing_key[2][1] == "abcdef0123456789"
+
+
+def test_pagerduty_from_mk_file_format_normalizes_legacy_explicit_tuple() -> None:
+    """Legacy on-disk values written by 2.4.0 REST API as
+    `("routing_key", key)` must be normalised on read so the API can return
+    them without raising and so they get rewritten in the new format.
+    """
+    from cmk.gui.rest_api_types.notifications_types import PagerDutyPlugin
+
+    legacy_params = {
+        "routing_key": ("routing_key", "abcdef0123456789"),
+        "webhook_url": "https://events.pagerduty.com/v2/enqueue",
+    }
+
+    plugin = PagerDutyPlugin.from_mk_file_format(legacy_params)
+    _, params = plugin.to_mk_file_format()
+    assert params is not None
+    routing_key = params["routing_key"]
+    assert routing_key[0] == "cmk_postprocessed"
+    assert routing_key[1] == "explicit_password"
+    assert routing_key[2][1] == "abcdef0123456789"
+
+
+def test_pagerduty_from_mk_file_format_normalizes_legacy_store_tuple() -> None:
+    """Legacy on-disk values for password-store-backed keys must be normalised."""
+    from cmk.gui.rest_api_types.notifications_types import PagerDutyPlugin
+
+    legacy_params = {
+        "routing_key": ("store", "my_pd_key"),
+        "webhook_url": "https://events.pagerduty.com/v2/enqueue",
+    }
+
+    plugin = PagerDutyPlugin.from_mk_file_format(legacy_params)
+    _, params = plugin.to_mk_file_format()
+    assert params is not None
+    routing_key = params["routing_key"]
+    assert routing_key[0] == "cmk_postprocessed"
+    assert routing_key[1] == "stored_password"
+    assert routing_key[2][0] == "my_pd_key"
+
+
+def test_pagerduty_from_mk_file_format_preserves_current_format() -> None:
+    """Values already in the current 3-tuple form must round-trip unchanged."""
+    from cmk.gui.rest_api_types.notifications_types import PagerDutyPlugin
+
+    current_params = {
+        "routing_key": (
+            "cmk_postprocessed",
+            "explicit_password",
+            ("uuid-1234", "abcdef0123456789"),
+        ),
+        "webhook_url": "https://events.pagerduty.com/v2/enqueue",
+    }
+
+    plugin = PagerDutyPlugin.from_mk_file_format(current_params)
+    _, params = plugin.to_mk_file_format()
+    assert params is not None
+    assert params["routing_key"] == current_params["routing_key"]