quick_setup: Enforce permission check

This change allows permission check for Quick Setup actions

* Allows to specify a required permission on the action level
* If required permission is set, then validate it during Quick Setup
  requests
* Added permissions required on AWS, Azure and GCP Quick Setups
* Enforce the permissions of the Quick Setup endpoints according to the
  permissions declared in the actions.

CMK-27975

Change-Id: I963d460deac564dadaea25c56940998cc5ebefb4

Author: lpetrora

cmk/gui/openapi/endpoints/quick_setup/__init__.py

@@ -63,6 +63,7 @@
 from cmk.gui.quick_setup.v0_unstable.predefined import build_formspec_map_from_stages
 from cmk.gui.quick_setup.v0_unstable.predefined._common import _find_id_in_form_data
 from cmk.gui.quick_setup.v0_unstable.setups import (
+    get_all_permissions,
     QuickSetupActionMode,
     QuickSetupBackgroundAction,
     QuickSetupBackgroundStageAction,
@@ -151,6 +152,7 @@ class QuickSetupMode(StrEnum):
     query_params=[QUICKSETUP_MODE, QUICKSETUP_OBJECT_ID],
     path_params=[QUICKSETUP_ID],
     response_schema=QuickSetupResponse,
+    additional_status_codes=[403],
 )
 def get_guided_stages_or_overview_stages(params: Mapping[str, Any]) -> Response:
     """Get guided stages or overview stages"""
@@ -166,6 +168,15 @@ def get_guided_stages_or_overview_stages(params: Mapping[str, Any]) -> Response:
             detail=f"Quick setup with id '{quick_setup_id}' does not exist.",
         )
 
+    permissions = get_all_permissions(quick_setup)
+
+    if permissions is not None and not all(user.may(perm) for perm in permissions):
+        return _serve_error(
+            title="Action not allowed",
+            detail=f"Requires {', '.join(f"'{p}'" for p in permissions)} permissions.",
+            status_code=403,
+        )
+
     mode: QuickSetupMode = params["mode"]
     prefill_data: ParsedFormData | None = None
     if object_id := params["object_id"]:
@@ -205,6 +216,7 @@ def get_guided_stages_or_overview_stages(params: Mapping[str, Any]) -> Response:
     path_params=[QUICKSETUP_ID, STAGE_INDEX],
     query_params=[QUICKSETUP_OBJECT_ID],
     response_schema=QuickSetupStageStructure,
+    additional_status_codes=[403],
 )
 def quick_setup_get_stage_structure(params: Mapping[str, Any]) -> Response:
     """Get a Quick setup stage structure"""
@@ -220,6 +232,15 @@ def quick_setup_get_stage_structure(params: Mapping[str, Any]) -> Response:
             detail=f"Quick setup with id '{quick_setup_id}' does not exist.",
         )
 
+    permissions = get_all_permissions(quick_setup)
+
+    if permissions is not None and not all(user.may(perm) for perm in permissions):
+        return _serve_error(
+            title="Action not allowed",
+            detail=f"Requires {', '.join(f"'{p}'" for p in permissions)} permissions.",
+            status_code=403,
+        )
+
     prefill_data: ParsedFormData | None = None
     if object_id:
         prefill_data = quick_setup.load_data(object_id)
@@ -246,13 +267,20 @@ def quick_setup_get_stage_structure(params: Mapping[str, Any]) -> Response:
         303: "The stage validation and recap action has been started in the background. "
         "Redirecting to the 'Get background job status snapshot' endpoint."
     },
-    additional_status_codes=[303],
+    additional_status_codes=[303, 403],
     path_params=[QUICKSETUP_ID],
     request_schema=QuickSetupStageActionRequest,
     response_schema=QuickSetupStageActionResponse,
 )
 def quicksetup_run_stage_action(params: Mapping[str, Any]) -> Response:
-    """Run a Quick setup stage validation and recap action"""
+    """Run a Quick setup stage validation and recap action
+
+    This endpoint performs permission validation but since permissions depend on dynamic actions
+    and each action has its own required permission, they cannot be statically defined. If the
+    required permissions for an action are not met, the endpoint returns a 403 Forbidden error
+    informing the action ID and the missing permission.
+    """
+
     language = user.language
     body = params["body"]
     quick_setup_id = params["quick_setup_id"]
@@ -267,6 +295,15 @@ def quicksetup_run_stage_action(params: Mapping[str, Any]) -> Response:
     stage_index = StageIndex(len(body["stages"]) - 1)
     stage_action = matching_stage_action(quick_setup.stages[stage_index](), stage_action_id)
 
+    if stage_action.permissions is not None and not all(
+        user.may(perm) for perm in stage_action.permissions
+    ):
+        return _serve_error(
+            title="Action not allowed",
+            detail=f"Action with id '{stage_action_id}' requires {', '.join(f"'{x}'" for x in stage_action.permissions)} permissions.",
+            status_code=403,
+        )
+
     built_stages = [stage() for stage in quick_setup.stages[: stage_index + 1]]
     form_spec_map = build_formspec_map_from_stages(built_stages)
     stages_raw_formspecs = [RawFormData(stage["form_data"]) for stage in body["stages"]]
@@ -384,7 +421,7 @@ def fetch_quick_setup_stage_action_result(params: Mapping[str, Any]) -> Response
     tag_group="Checkmk Internal",
     path_params=[QUICKSETUP_ID],
     query_params=[QUICKSETUP_MODE],
-    additional_status_codes=[201, 303, 429],
+    additional_status_codes=[201, 303, 403, 429],
     status_descriptions={
         303: "The validation and complete action has been started in the background. "
         "Redirecting to the 'Get background job status snapshot' endpoint.",
@@ -405,7 +442,7 @@ def quick_setup_run_action(params: Mapping[str, Any]) -> Response:
     tag_group="Checkmk Internal",
     path_params=[QUICKSETUP_ID],
     query_params=[QUICKSETUP_OBJECT_ID_REQUIRED],
-    additional_status_codes=[201, 303, 429],
+    additional_status_codes=[201, 303, 403, 429],
     status_descriptions={
         303: "The validation and complete action has been started in the background. "
         "Redirecting to the 'Get background job status snapshot' endpoint.",
@@ -450,6 +487,13 @@ def complete_quick_setup_action(params: Mapping[str, Any], mode: QuickSetupActio
             detail=f"Action with id '{action_id}' does not exist.",
         )
 
+    if action.permissions is not None and not all(user.may(perm) for perm in action.permissions):
+        return _serve_error(
+            title="Action not allowed",
+            detail=f"Action with id '{action_id}' requires {', '.join(f"'{x}'" for x in action.permissions)} permissions.",
+            status_code=403,
+        )
+
     form_spec_map = build_formspec_map_from_stages([stage() for stage in quick_setup.stages])
     errors = validate_stages_form_data(
         stages_raw_form_data=[RawFormData(stage["form_data"]) for stage in body["stages"]],

cmk/gui/quick_setup/config_setups/aws/stages.py

@@ -132,6 +132,7 @@ def prepare_aws() -> QuickSetupStage:
                 ],
                 recap=[recaps.recaps_form_spec],
                 next_button_label=_("Configure host and regions"),
+                permissions=["wato.passwords"],
             )
         ],
     )
@@ -163,6 +164,7 @@ def configure_host_and_regions() -> QuickSetupStage:
                 ],
                 recap=[recaps.recaps_form_spec],
                 next_button_label=_("Configure services to monitor"),
+                permissions=["wato.hosts"],
             )
         ],
         prev_button_label=PREV_BUTTON_LABEL,
@@ -445,6 +447,7 @@ def aws_transform_to_disk(params: Mapping[str, object]) -> Mapping[str, object]:
             action=action,
             icon=QuickSetupActionButtonIcon(name="save-to-services"),
             custom_validators=[qs_validators.validate_host_name_doesnt_exists],
+            permissions=["wato.passwords", "wato.rulesets", "wato.hosts"],
         ),
     ],
 )

cmk/gui/quick_setup/config_setups/azure/stages.py

@@ -155,6 +155,7 @@ def configure_authentication() -> QuickSetupStage:
                 ],
                 recap=[recaps.recaps_form_spec],
                 next_button_label=_("Configure host and authority"),
+                permissions=["wato.passwords"],
             ),
         ],
     )
@@ -179,6 +180,7 @@ def configure_host_and_authority() -> QuickSetupStage:
                 ],
                 recap=[recaps.recaps_form_spec],
                 next_button_label=_("Configure services to monitor"),
+                permissions=["wato.hosts"],
             ),
         ],
         prev_button_label=PREV_BUTTON_LABEL,
@@ -356,6 +358,7 @@ def azure_collect_params(
             label=_("Save & go to Activate changes"),
             icon=QuickSetupActionButtonIcon(name="save-to-services"),
             action=action,
+            permissions=["wato.passwords", "wato.rulesets", "wato.hosts"],
         ),
     ],
 )

cmk/gui/quick_setup/config_setups/gcp/stages.py

@@ -130,6 +130,7 @@ def configure_authentication() -> QuickSetupStage:
                 ],
                 recap=[recaps.recaps_form_spec],
                 next_button_label=_("Configure host"),
+                permissions=["wato.passwords"],
             )
         ],
     )
@@ -152,6 +153,7 @@ def configure_host() -> QuickSetupStage:
                 ],
                 recap=[recaps.recaps_form_spec],
                 next_button_label=_("Configure services to monitor"),
+                permissions=["wato.hosts"],
             )
         ],
         prev_button_label=PREV_BUTTON_LABEL,
@@ -345,6 +347,7 @@ def gcp_collect_params(
             label=_("Save & go to Activate changes"),
             icon=QuickSetupActionButtonIcon(name="save-to-services"),
             action=action,
+            permissions=["wato.passwords", "wato.rulesets", "wato.hosts"],
         ),
     ],
 )

cmk/gui/quick_setup/v0_unstable/setups.py

@@ -83,13 +83,18 @@ class QuickSetupStageAction:
 
             load_wait_label:
                 The label of the loading spinner. If not set, the default label is used.
+
+            permissions:
+                The required permission name to be checked by the Quick Setup endpoint. By default
+                it is set to None and no permission check is performed
     """
 
     id: ActionId
     custom_validators: Iterable[CallableValidator]
     recap: Iterable[CallableRecap]
     next_button_label: str | None = None
     load_wait_label: str | None = None
+    permissions: list[str] | None = None
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -181,13 +186,17 @@ class QuickSetupAction:
 
             Therefore, relevant custom validators should be included again here (especially
             if the 'overview' mode is enabled)
+        permissions:
+            The required permission name to be checked by the Quick Setup endpoint. By default
+            it is set to None and no permission check is performed
     """
 
     id: ActionId
     label: str
     action: CallableAction
     icon: QuickSetupActionButtonIcon | None = None
     custom_validators: Iterable[CallableValidator] = ()
+    permissions: list[str] | None = None
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -205,3 +214,17 @@ class QuickSetup:
     stages: Sequence[Callable[[], QuickSetupStage]]
     actions: Sequence[QuickSetupAction]
     load_data: Callable[[str], ParsedFormData | None] = lambda _: None
+
+
+def get_all_permissions(quick_setup: QuickSetup) -> list[str] | None:
+    action_sources = [stage() for stage in quick_setup.stages] + [quick_setup]
+    permissions = list(
+        {
+            perm
+            for source in action_sources
+            for action in source.actions
+            if action.permissions
+            for perm in action.permissions
+        }
+    )
+    return permissions or None

tests/unit/cmk/gui/openapi/test_openapi_quick_setup.py

@@ -743,3 +743,73 @@ def test_openapi_background_job_setup_action(self, clients: ClientRegistry) -> N
             expect_ok=True,
         )
         assert "objects/background_job" in resp.headers["Location"]
+
+
+@pytest.mark.parametrize(
+    "perm",
+    [(["general.use"]), (None)],
+)
+def test_validate_permission(clients: ClientRegistry, perm: list[str] | None) -> None:
+    register_quick_setup(
+        setup_stages=[
+            lambda: QuickSetupStage(
+                title="stage1",
+                configure_components=[
+                    widgets.unique_id_formspec_wrapper(Title("account name")),
+                ],
+                actions=[
+                    QuickSetupStageAction(
+                        id=ActionId("action"),
+                        custom_validators=[],
+                        recap=[recaps.recaps_form_spec],
+                        next_button_label="Next",
+                        permissions=perm,
+                    )
+                ],
+            ),
+        ],
+    )
+    res = clients.QuickSetup.run_stage_action(
+        quick_setup_id="quick_setup_test",
+        stage_action_id="action",
+        stages=[{"form_data": {UniqueFormSpecIDStr: {UniqueBundleIDStr: "test_account_name"}}}],
+    )
+    assert res.json["validation_errors"] is None
+    assert res.json["background_job_exception"] is None
+    assert res.json["stage_recap"][0]["id"] == "formspec_unique_id"
+    assert res.json["stage_recap"][0]["form_spec"]["data"]["bundle_id"] == "test_account_name"
+
+
+def test_fail_validate_permission(clients: ClientRegistry) -> None:
+    register_quick_setup(
+        setup_stages=[
+            lambda: QuickSetupStage(
+                title="stage1",
+                configure_components=[
+                    widgets.unique_id_formspec_wrapper(Title("account name")),
+                ],
+                actions=[
+                    QuickSetupStageAction(
+                        id=ActionId("action"),
+                        custom_validators=[],
+                        recap=[recaps.recaps_form_spec],
+                        next_button_label="Next",
+                        permissions=["impossible_permisson"],
+                    )
+                ],
+            ),
+        ],
+    )
+    resp = clients.QuickSetup.run_stage_action(
+        quick_setup_id="quick_setup_test",
+        stage_action_id="action",
+        stages=[{"form_data": {UniqueFormSpecIDStr: {UniqueBundleIDStr: "test_account_name"}}}],
+        expect_ok=False,
+    )
+
+    assert resp.status_code == 403
+    assert resp.json["title"] == "Action not allowed"
+    assert (
+        resp.json["detail"]
+        == "Action with id 'action' requires 'impossible_permisson' permissions."
+    )