attempt to handle module validation

sekoia_automation/module.py

@@ -1,3 +1,4 @@
+import ast
 import json
 import logging
 import sys
@@ -325,6 +326,7 @@ class ModuleItem(ABC):
     SECRETS_URL_FILE_NAME = "url_secrets"
     LOGS_URL_FILE_NAME = "url_logs"
     INTAKE_URL_FILE_NAME = "intake_url"
+    VALIDATION_CALLBACK_URL_FILE_NAME = "validation_callback_url"
 
     name: str | None = None
     description: str | None = None
@@ -487,3 +489,98 @@ def stop_monitoring(self):
         """
         Stops the background monitoring operations
         """
+
+        def validate(self):
+        # Check if "validate_module_configuration" is in the docker parameters
+        main_py_path = self.__module_path / "main.py"
+        with open(main_py_path) as file:
+            content = file.read()
+
+        tree = ast.parse(content)
+
+        docker_params = []
+        node: Any
+        for node in ast.walk(tree):
+            if (
+                hasattr(node, "func")
+                and isinstance(node.func, ast.Attribute)
+                and node.func.attr == "register"
+            ):
+                docker_param: str | None = None
+                if len(node.args) > 1 and isinstance(node.args[1], ast.Constant):
+                    # provided as positional arg
+                    docker_param = node.args[1].s
+
+                elif len(node.keywords) > 0:
+                    # provided as keyword arg
+                    docker_param = node.keywords[0].value.s
+
+                docker_params.append(docker_param)
+
+        if "validate_module_configuration" not in docker_params:
+            return
+
+        # Call the actual validation procedure
+        status = self.do_validation()
+
+        # Return result of validation to Symphony
+        data = {"validation_status": status}
+        validation_callback_url = self.load_config(
+            self.VALIDATION_CALLBACK_URL_FILE_NAME
+        )
+        response = requests.request(
+            "POST",
+            validation_callback_url,
+            json=data,
+            headers={"Authorization": f"Bearer {self.token}"},
+            timeout=30,
+        )
+        response.raise_for_status()
+
+    def validate(self):
+        # Check if "validate_module_configuration" is in the docker parameters
+        main_py_path = self.__module_path / "main.py"
+        with open(main_py_path) as file:
+            content = file.read()
+
+        tree = ast.parse(content)
+
+        docker_params = []
+        node: Any
+        for node in ast.walk(tree):
+            if (
+                hasattr(node, "func")
+                and isinstance(node.func, ast.Attribute)
+                and node.func.attr == "register"
+            ):
+                docker_param: str | None = None
+                if len(node.args) > 1 and isinstance(node.args[1], ast.Constant):
+                    # provided as positional arg
+                    docker_param = node.args[1].s
+
+                elif len(node.keywords) > 0:
+                    # provided as keyword arg
+                    docker_param = node.keywords[0].value.s
+
+                docker_params.append(docker_param)
+
+        if "validate_module_configuration" not in docker_params:
+            return
+
+        # Call the actual validation procedure
+        status = self.do_validation()
+
+        # Return result of validation to Symphony
+        data = {"validation_status": status}
+        validation_callback_url = self.module.load_config(
+            self.VALIDATION_CALLBACK_URL_FILE_NAME
+        )
+        response = requests.request(
+            "POST",
+            validation_callback_url,
+            json=data,
+            headers={"Authorization": f"Bearer {self.token}"},
+            timeout=30,
+        )
+        response.raise_for_status()
+