Refactor automation page to use DistributedSetupSecret

CMK-16715

Change-Id: I8cc225b7c89a330fc5d0d43b209450e7b69b0224

cmk/gui/wato/pages/automation.py

@@ -17,6 +17,7 @@
 from cmk.ccc.site import omd_site
 
 import cmk.utils.paths
+from cmk.utils.local_secrets import DistributedSetupSecret
 from cmk.utils.paths import configuration_lockfile
 from cmk.utils.user import UserId
 
@@ -42,6 +43,7 @@
 )
 
 from cmk import trace
+from cmk.crypto.password import Password
 
 tracer = trace.get_tracer()
 
@@ -85,7 +87,7 @@ def page(self) -> PageResult:  # pylint: disable=useless-return
                 {
                     "version": cmk_version.__version__,
                     "edition_short": cmk_version.edition(cmk.utils.paths.omd_root).short,
-                    "login_secret": _get_login_secret(create_on_demand=True),
+                    "login_secret": DistributedSetupSecret().read_or_create(),
                 }
             )
         )
@@ -111,14 +113,11 @@ def _from_vars(self) -> None:
 
     @staticmethod
     def _authenticate() -> None:
-        secret = request.var("secret")
-
+        secret = request.get_validated_type_input(Password, "secret")
         if not secret:
             raise MKAuthException(_("Missing secret for automation command."))
 
-        login_secret = _get_login_secret()
-
-        if (login_secret is None) or not secrets.compare_digest(secret, login_secret):
+        if not DistributedSetupSecret().compare(secret):
             raise MKAuthException(_("Invalid automation secret."))
 
     # TODO: Better use AjaxPage.handle_page() for standard AJAX call error handling. This

cmk/utils/local_secrets.py

@@ -8,9 +8,12 @@
 import secrets
 from pathlib import Path
 
+from cmk.ccc import store
+
 from cmk.utils import paths
 from cmk.utils.user import UserId
 
+from cmk.crypto.password import Password
 from cmk.crypto.secrets import LocalSecret, Secret
 
 
@@ -101,3 +104,37 @@ def delete(self) -> None:
     def check(self, other: str) -> bool:
         """Check if a given secret is the same as this one in a timing attack safe manner"""
         return secrets.compare_digest(self.read().encode("utf-8"), other.encode("utf-8"))
+
+
+class DistributedSetupSecret:
+    """The secret used by the central site to sync the config"""
+
+    def __init__(self) -> None:
+        self.password = (
+            Password(pw)
+            if (pw := store.load_object_from_file(self.path, default=None)) is not None
+            else None
+        )
+
+    @property
+    def path(self) -> Path:
+        return Path(paths.var_dir) / "wato" / "automation_secret.mk"
+
+    def read_or_create(self) -> Password:
+        if self.password is None:
+            self.regenerate()
+        assert self.password is not None
+        return self.password
+
+    def regenerate(self) -> None:
+        """Generate a new secret and write it to disk.
+
+        If the file already exists, it will be overwritten.
+        """
+        self.password = Password(secrets.token_urlsafe(32))
+        store.save_object_to_file(self.path, self.password.raw)
+
+    def compare(self, other: Password) -> bool:
+        if self.password is None:
+            return False
+        return self.password == other

tests/unit/cmk/gui/wato/pages/test_automation.py

@@ -13,6 +13,8 @@
 import cmk.ccc.version as cmk_version
 from cmk.ccc.exceptions import MKGeneralException
 
+from cmk.utils.local_secrets import DistributedSetupSecret
+
 from cmk.automations.results import ABCAutomationResult, ResultTypeRegistry, SerializedResult
 
 from cmk.gui.exceptions import MKAuthException
@@ -72,7 +74,9 @@ def patch_edition_fixture(self, monkeypatch: pytest.MonkeyPatch) -> None:
 
     @pytest.fixture(name="fix_secret_checking")
     def patch_distributed_setup_secret(self, monkeypatch: pytest.MonkeyPatch) -> None:
-        monkeypatch.setattr(automation, "_get_login_secret", lambda **_kwargs: "secret")
+        monkeypatch.setattr(
+            DistributedSetupSecret, "compare", lambda _self, other: other.raw == "secret"
+        )
 
     @pytest.fixture(name="setup_request")
     def setup_request_fixture(self, monkeypatch: pytest.MonkeyPatch) -> None: