Add account validation module

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## 1.17.0 - 2024-11-04
+
+### Added
+
+- Add account validation (beta)
+
 ## 1.16.1 - 2024-10-30
 
 ### Changed

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry]
 name = "sekoia-automation-sdk"
 
-version = "1.16.1"
+version = "1.17.0"
 description = "SDK to create Sekoia.io playbook modules"
 license = "MIT"
 readme = "README.md"

sekoia_automation/account_validator.py

@@ -0,0 +1,50 @@
+from abc import abstractmethod
+from pathlib import Path
+
+from sekoia_automation.module import Module, ModuleItem
+
+
+class AccountValidator(ModuleItem):
+    CALLBACK_URL_FILE_NAME = "validation_callback_url"
+
+    def __init__(self, module: Module | None = None, data_path: Path | None = None):
+        super().__init__(module, data_path)
+        self._error: str | None = None
+
+    @abstractmethod
+    def validate(self) -> bool:
+        """To define in subclasses. Validates the configuration of the module.
+
+        Returns:
+            bool: True if the module is valid, False otherwise
+        """
+
+    def error(self, message: str) -> None:
+        """Allow to set an error message explaining why the validation failed."""
+        self._error = message
+
+    def execute(self):
+        """Validates the account (module_configuration) of the module
+        and sends the result to Symphony."""
+        self.set_task_as_running()
+        # Call the actual validation procedure
+        success = self.validate()
+        self.send_results(success)
+
+    def set_task_as_running(self):
+        """Send a request to indicate the action started."""
+        data = {"status": "running"}
+        response = self._send_request(data, verb="PATCH")
+        if self.module.has_secrets():
+            secrets = {
+                k: v
+                for k, v in response.json()["module_configuration"]["value"].items()
+                if k in self.module.manifest_secrets()
+            }
+            self.module.set_secrets(secrets)
+
+    def send_results(self, success: bool):
+        data = {"status": "finished", "results": {"success": success}}
+        if self._error:
+            data["error"] = self._error
+        self._send_request(data, verb="PATCH")

sekoia_automation/module.py

@@ -5,7 +5,7 @@
 from abc import ABC, abstractmethod
 from functools import cached_property
 from pathlib import Path
-from typing import Any, Literal, cast
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 import requests
 import sentry_sdk
@@ -25,6 +25,10 @@
     get_as_model,
 )
 
+if TYPE_CHECKING:  # pragma: no cover
+    from sekoia_automation.account_validator import AccountValidator
+
+
 LogLevelStr = Literal["fatal", "critical", "error", "warning", "info", "debug"]
 
 
@@ -269,6 +273,9 @@ def register(self, item: type["ModuleItem"], name: str = ""):
             item.name = name
         self._items[name] = item
 
+    def register_account_validator(self, validator: type["AccountValidator"]):
+        self.register(validator, "validate_module_configuration")
+
     def run(self):
         command = self.command or ""
 

tests/test_account_validator.py

@@ -0,0 +1,112 @@
+from unittest.mock import patch
+
+import requests_mock
+
+from sekoia_automation.account_validator import AccountValidator
+from sekoia_automation.module import Module
+
+
+class MockAccountValidator(AccountValidator):
+    mock_return_value = True
+
+    def validate(self):
+        if not self.mock_return_value:
+            self.error("Validation failed")
+        return self.mock_return_value
+
+
+def test_execute_success():
+    validator = MockAccountValidator()
+    validator.mock_return_value = True
+
+    with (
+        patch.object(
+            validator.module, "load_config", return_value="http://example.com/callback"
+        ),
+        requests_mock.Mocker() as mock_request,
+    ):
+        mock_request.patch("http://example.com/callback", status_code=200)
+
+        validator.execute()
+
+        # Check the callback has been called
+        assert mock_request.call_count == 2
+        assert mock_request.request_history[0].json() == {"status": "running"}
+        assert mock_request.last_request.json() == {
+            "results": {"success": True},
+            "status": "finished",
+        }
+
+
+def test_execute_failure():
+    validator = MockAccountValidator()
+    validator.mock_return_value = False
+
+    with (
+        patch.object(
+            validator.module, "load_config", return_value="http://example.com/callback"
+        ),
+        requests_mock.Mocker() as mock_request,
+    ):
+        mock_request.patch("http://example.com/callback", status_code=200)
+
+        validator.execute()
+
+        # Check the callback has been called
+        assert mock_request.call_count == 2
+        assert mock_request.request_history[0].json() == {"status": "running"}
+        assert mock_request.last_request.json() == {
+            "error": "Validation failed",
+            "results": {"success": False},
+            "status": "finished",
+        }
+
+
+def test_execute_with_secrets():
+    module = Module()
+    module._manifest = {
+        "configuration": {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "properties": {
+                "api_key": {"description": "SEKOIA.IO API key", "type": "string"},
+                "base_url": {
+                    "description": "SEKOIA.IO base URL (ex. https://api.sekoia.io)",
+                    "type": "string",
+                },
+            },
+            "required": ["api_key"],
+            "secrets": ["api_key"],
+            "title": "SEKOIA.IO Configuration",
+            "type": "object",
+        }
+    }
+    module._configuration = {"base_url": "https://api.sekoia.io"}
+    validator = MockAccountValidator(module=module)
+    validator.mock_return_value = True
+
+    with (
+        patch.object(
+            validator.module, "load_config", return_value="http://example.com/callback"
+        ),
+        requests_mock.Mocker() as mock_request,
+    ):
+        mock_request.patch(
+            "http://example.com/callback",
+            status_code=200,
+            json={"module_configuration": {"value": {"api_key": "foo"}}},
+        )
+
+        validator.execute()
+
+        # Check the configuration has been updated with the secrets
+        assert module.configuration == {
+            "api_key": "foo",
+            "base_url": "https://api.sekoia.io",
+        }
+        # Check the callback has been called
+        assert mock_request.call_count == 2
+        assert mock_request.request_history[0].json() == {"status": "running"}
+        assert mock_request.last_request.json() == {
+            "results": {"success": True},
+            "status": "finished",
+        }

tests/test_module.py

@@ -1,5 +1,5 @@
 # natives
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 # third parties
 import pytest
@@ -63,6 +63,14 @@ def test_register_no_command():
             module.run()
 
 
+def test_register_account_validator():
+    module = Module()
+    validator = Mock()
+    validator.name = None
+    module.register_account_validator(validator)
+    assert module._items["validate_module_configuration"] == validator
+
+
 @patch.object(DummyTrigger, "execute")
 def test_register_execute_default(mock):
     module = Module()