Azure quick setup: connection validation

introduce a new Azure agent option "--connection-test" and use it
in the quick setup stage validation

CMK-19762

Change-Id: I91d310c2a1a9c3c8479ba4e745b529d152dff2ff

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

@@ -52,6 +52,18 @@
 PREV_BUTTON_LABEL = _("Back")
 
 
+def _collect_params_for_connection_test(
+    all_stages_form_data: ParsedFormData, parameter_form: Dictionary
+) -> Mapping[str, object]:
+    """For the quick setup validation of the Azure authentication we run a connection test only.
+    The agent option "--connection-test" is added, running only a connection via the Management API
+    client (none via the Graph API client)."""
+    return {
+        **collect_params_with_defaults_from_form_data(all_stages_form_data, parameter_form),
+        "connection_test": True,
+    }
+
+
 def configure_authentication() -> QuickSetupStage:
     return QuickSetupStage(
         title=_("Prepare Azure for Checkmk"),
@@ -103,7 +115,7 @@ def configure_authentication() -> QuickSetupStage:
             qs_validators.validate_test_connection_custom_collect_params(
                 rulespec_name=RuleGroup.SpecialAgents("azure"),
                 parameter_form=azure.formspec(),
-                custom_collect_params=collect_params_with_defaults_from_form_data,
+                custom_collect_params=_collect_params_for_connection_test,
                 error_message=_(
                     "Could not access your Azure account. Please check your credentials."
                 ),

cmk/plugins/azure/server_side_calls/agent_azure.py

@@ -46,6 +46,7 @@ class AzureParams(BaseModel):
     config: Config
     piggyback_vms: str | None = None
     import_tags: tuple[str, str | None] | None = None
+    connection_test: bool = False  # only used by quick setup
 
 
 def _tag_based_args(tag_based: list[TagBased]) -> list[str]:
@@ -79,7 +80,10 @@ def agent_azure_arguments(
         params.secret.unsafe(),
     ]
     if params.authority:
-        args += ["--authority", params.authority if params.authority != "global_" else "global"]
+        args += [
+            "--authority",
+            params.authority if params.authority != "global_" else "global",
+        ]
     if params.subscription:
         args += ["--subscription", params.subscription]
     if params.piggyback_vms:
@@ -110,6 +114,9 @@ def agent_azure_arguments(
     if config.tag_based:
         args += _tag_based_args(config.tag_based)
 
+    if params.connection_test:
+        args += ["--connection-test"]
+
     yield SpecialAgentCommand(command_arguments=args)
 
 

cmk/special_agents/agent_azure.py

@@ -356,6 +356,14 @@ def parse_arguments(argv: Sequence[str]) -> Args:
         "for in the key of the tag.",
     )
     group_import_tags.set_defaults(tag_key_pattern=TagsImportPatternOption.import_all)
+
+    parser.add_argument(
+        "--connection-test",
+        action="store_true",
+        help="Run a connection test through the Management API only. No further agent code is "
+        "executed.",
+    )
+
     args = parser.parse_args(argv)
 
     # LOGGING
@@ -1875,6 +1883,26 @@ def process_resource_health(
         yield section
 
 
+def test_connection(args: Args, subscription: str) -> int | tuple[int, str]:
+    """We test the connection only via the Management API client, not via the Graph API client.
+    The Graph API client is used for three specific services, which are disabled in the default
+    setup when configured via the UI.
+    The Management API client is used for all other services, so we assume here that this is the
+    connection that's essential for the vast majority of setups."""
+    mgmt_client = MgmtApiClient(
+        _get_mgmt_authority_urls(args.authority, subscription),
+        deserialize_http_proxy_config(args.proxy),
+        subscription,
+    )
+    try:
+        mgmt_client.login(args.tenant, args.client, args.secret)
+    except ValueError as exc:
+        error_msg = f"Management client login failed with: {exc}\n"
+        sys.stdout.write(error_msg)
+        return 2, error_msg
+    return 0
+
+
 def main_subscription(args: Args, selector: Selector, subscription: str) -> None:
     mgmt_client = MgmtApiClient(
         _get_mgmt_authority_urls(args.authority, subscription),
@@ -1884,7 +1912,6 @@ def main_subscription(args: Args, selector: Selector, subscription: str) -> None
 
     try:
         mgmt_client.login(args.tenant, args.client, args.secret)
-
         all_resources = (AzureResource(r, args.tag_key_pattern) for r in mgmt_client.resources())
 
         monitored_resources = [r for r in all_resources if selector.do_monitor(r)]
@@ -1926,12 +1953,18 @@ def main(argv=None):
     selector = Selector(args)
     if args.dump_config:
         sys.stdout.write("Configuration:\n%s\n" % selector)
-        return
+        return 0
+
+    if args.connection_test:
+        for subscription in args.subscriptions:
+            if (test_result := test_connection(args, subscription)) != 0:
+                return test_result
 
     LOGGER.debug("%s", selector)
     main_graph_client(args)
     for subscription in args.subscriptions:
         main_subscription(args, selector, subscription)
+    return 0
 
 
 if __name__ == "__main__":

tests/unit/cmk/special_agents/agent_azure/test_agent_azure_config.py

@@ -62,6 +62,7 @@
     explicit_config=["group=test-group", "resources=Resource1,Resource2"],
     services=["Microsoft.Compute/virtualMachines", "Microsoft.Storage/storageAccounts"],
     tag_key_pattern=TagsImportPatternOption.import_all,
+    connection_test=False,
 )
 
 
@@ -89,12 +90,16 @@
                 "argparse: services = ['Microsoft.Compute/virtualMachines', 'Microsoft.Storage/storageAccounts']",
                 "argparse: authority = 'global'",
                 "argparse: tag_key_pattern = <TagsImportPatternOption.import_all: 'IMPORT_ALL'>",
+                "argparse: connection_test = False",
             ],
         ),
     ],
 )
 def test_parse_arguments(
-    argv: Sequence[str], args: Args, expected_log: Sequence[str], caplog: pytest.LogCaptureFixture
+    argv: Sequence[str],
+    args: Args,
+    expected_log: Sequence[str],
+    caplog: pytest.LogCaptureFixture,
 ) -> None:
     caplog.set_level(logging.DEBUG)
     assert parse_arguments(argv) == args