Implement Mandatory and Optional fields functionality

docs/DocGenerator/CodeParser.py

@@ -5,10 +5,14 @@
 import yaml
 from Config import Config
 from TestCaseParser import TestCaseParser
+from Utils import remove_inexistent
 from docstring_parser import parse
 from comment_parser import comment_parser
 import warnings
 
+INTERNAL_FIELDS = ['Id', 'Group Id', 'Name']
+STOP_FIELDS = ['Tests','Test Cases']
+
 class CodeParser:
     def __init__(self):
         self.conf = Config()
@@ -24,23 +28,25 @@ def is_documentable_function(self, function):
         return False
 
     def remove_ignored_fields(self, doc):
-        for field in self.conf.ignored_fields.module:
-            if field in doc:
-                del doc[field]
-        for test in doc['Tests']:
-            for field in self.conf.ignored_fields.tests:
-                if field in test:
-                    del test[field]
+        allowed_fields = self.conf.module_fields.mandatory + self.conf.module_fields.optional + INTERNAL_FIELDS
+        remove_inexistent(doc, allowed_fields, STOP_FIELDS)
+        if 'Tests' in doc:
+            allowed_fields = self.conf.test_fields.mandatory + self.conf.test_fields.optional + INTERNAL_FIELDS
+            for test in doc['Tests']:
+                remove_inexistent(test, allowed_fields, STOP_FIELDS)
 
     def parse_comment(self, function):
         docstring = ast.get_docstring(function)
         try:
-            doc = yaml.load(docstring)
+            doc = yaml.safe_load(docstring)
             if hasattr(function, 'name'):
                 doc['Name'] = function.name
 
         except Exception as inst:
-            warnings.warn("Error parsing comment of...")
+            if hasattr(function, 'name'):
+                warnings.warn(f"Error parsing comment of function '{function.name}'' from module {self.scan_file}")
+            else:
+                warnings.warn(f"Error parsing comment of module {self.scan_file}")
             print(type(inst))
             print(inst.args)
             print(inst)
@@ -49,6 +55,7 @@ def parse_comment(self, function):
         return doc
 
     def parse_test(self, code_file, id, group_id):
+        self.scan_file = code_file
         with open(code_file) as fd:
             file_content = fd.read()
         module = ast.parse(file_content)
@@ -76,7 +83,7 @@ def parse_test(self, code_file, id, group_id):
 
             module_doc['Tests'] = functions_doc
 
-            #self.remove_ignored_fields(module_doc)
+            self.remove_ignored_fields(module_doc)
 
         return module_doc
 

docs/DocGenerator/Sanity.py

@@ -3,7 +3,7 @@
 import re
 import json
 import ast
-
+from Utils import check_existance
 class Sanity():
     def __init__(self,):
         self.conf = Config()
@@ -22,15 +22,20 @@ def get_content(self, full_path):
             raise Exception(f"Cannot load {full_path} file")
 
     def validate_fields(self, required_fields, available_fields):
-        for field in required_fields:
-            if isinstance(field, dict):
-                for key in field:
-                    if key in available_fields:
-                        self.validate_fields(field[key], available_fields[key])
-                    else:
-                        self.add_report(f"Mandatory field '{key}' is missing in file {self.scan_file}")
-            elif not field in available_fields:
-                self.add_report(f"Mandatory field '{field}' is missing in file {self.scan_file}")
+        if isinstance(required_fields, dict):
+            for field in required_fields:
+                if not check_existance(available_fields, field):
+                    self.add_report(f"Mandatory field '{field}' is missing in file {self.scan_file}")
+                elif isinstance(required_fields[field], dict) or  isinstance(required_fields[field], list):
+                    self.validate_fields(required_fields[field], available_fields)
+        elif isinstance(required_fields, list):
+            for field in required_fields:
+                if isinstance(field, dict) or isinstance(field, list):
+                    self.validate_fields(field, available_fields)
+                else:
+                    if not check_existance(available_fields, field):
+                        self.add_report(f"Mandatory field '{field}' is missing in file {self.scan_file}")
+
 
     def validate_module_fields(self, fields):
         self.validate_fields(self.conf.module_fields.mandatory, fields)
@@ -40,7 +45,7 @@ def validate_test_fields(self, fields):
             self.validate_fields(self.conf.test_fields.mandatory, test_fields)
 
     def identify_tags(self, content):
-        if 'Tags' in content['Metadata']:
+        if 'Metadata' in content and 'Tags' in content['Metadata']:
             for tag in content['Metadata']['Tags']:
                 self.found_tags.add(tag)
 

docs/DocGenerator/Utils.py

@@ -0,0 +1,100 @@
+def check_existance(source, key):
+    if not isinstance(source, dict) and not isinstance(source, list):
+        return False
+
+    if key in source:
+        return True
+    elif isinstance(source, dict):
+        for item in source:
+            if check_existance(source[item], key):
+                return True
+        return False
+    elif isinstance(source, list):
+        for item in source:
+            if check_existance(item, key):
+                return True
+        return False
+    else:
+        return False
+
+def remove_inexistent(source, check_list, stop_list=None):
+    for element in list(source):
+        if stop_list and element in stop_list:
+            break
+        if not check_existance(check_list, element):
+            del source[element]
+        elif isinstance(source[element], dict):
+            remove_inexistent(source[element], check_list, stop_list)
+
+def get_keys_dict(dic):
+    keys = []
+    for item in dic:
+        value = dic[item]
+        if isinstance(value, dict):
+            result = get_keys_dict(value)
+            keys.append({item : result})
+        elif isinstance(value, list):
+            result = get_keys_list(value)
+            keys.append({item : result})
+        else:
+            keys.append(item)
+
+    if len(keys) == 1:
+        return keys[0]
+    else:
+        return keys
+
+def get_keys_list(dic):
+    keys = []
+    for item in dic:
+        if isinstance(item, dict):
+            result = get_keys_dict(item)
+            keys.append(result)
+        elif isinstance(item, list):
+            result = get_keys_list(item)
+            keys.append(result)
+        else:
+            keys.append(item)
+
+    if len(keys) == 1:
+        return keys[0]
+    else:
+        return keys
+
+def find_item(search_item, check):
+    for item in check:
+        if isinstance(item, dict):
+            list_element = list(item.keys())
+            if search_item == list_element[0]:
+                return list(item.values())[0]
+        else:
+            if search_item == item:
+                return item
+    return None
+
+def check_missing_field(source, check):
+    missing_filed = None
+    for source_field in source:
+        if isinstance(source_field, dict):
+            key = list(source_field.keys())[0]
+            found_item = find_item(key, check)
+            if not found_item:
+                print(f"Missing key {source_field}")
+                return key
+            missing_filed = check_missing_field(source_field[key], found_item)
+            if missing_filed:
+                return missing_filed
+        elif isinstance(source_field, list):
+            missing_filed = None
+            for check_element in check:
+                missing_filed = check_missing_field(source_field, check_element)
+                if not missing_filed:
+                    break
+            if missing_filed:
+                return source_field
+        else:
+            found_item = find_item(source_field, check)
+            if not found_item:
+                print(f"Missing key {source_field}")
+                return source_field
+    return missing_filed

docs/DocGenerator/config.yaml

@@ -6,8 +6,6 @@ Include paths:
   - wazuh_db:
     path: "../../tests/integration/test_wazuh_db"
     recursive: false
-  - vulnerability_detector:
-    path: "../../tests/integration/test_vulnerability_detector"
 
 Include regex:
   - "^test_.*py$"
@@ -22,12 +20,6 @@ Ignore paths:
   - "../../tests/integration/test_wazuh_db/data"
   - "/data/*"
 
-Valid tags:
-  - DB
-  - Feeds
-  - VulDet
-  - Vulnerability Detector
-
 Output fields:
   Module:
     Mandatory:
@@ -37,9 +29,6 @@ Output fields:
         - Daemons
         - Operating System
         - Tiers
-        - Demo:
-          - One
-          - Two
     Optional:
       - Tags
   Test:

tests/integration/test_wazuh_db/test_wazuh_db.py

@@ -1,6 +1,29 @@
-# Copyright (C) 2015-2021, Wazuh Inc.
-# Created by Wazuh, Inc. <info@wazuh.com>.
-# This program is free software; you can redistribute it and/or modify it under the terms of GPLv2
+'''
+Brief: Module description
+
+COPYRIGHT:
+    Copyright (C) 2015-2021, Wazuh Inc.
+
+    Created by Wazuh, Inc. <info@wazuh.com>.
+
+    This program is free software; you can redistribute it and/or modify it under the terms of GPLv2
+
+Metadata:
+    Modules:
+        - Wazuh DB
+    Daemons:
+        - wazuh_db
+    Operating System:
+        - Windows
+        - Ubuntu
+    Wazuh Max Version: 4.0.0
+    Wazuh Min Version: 4.1.5
+    Tiers:
+        - 0
+        - 1
+    Tags:
+        - Enrollment
+'''
 
 import os
 import re
@@ -83,13 +106,15 @@ def pre_insert_agents():
                               for case in module_data]
                          )
 def test_wazuh_db_messages(configure_sockets_environment, connect_to_sockets_module, test_case: list):
-    """Check that every input message in wazuh-db socket generates the adequate output to wazuh-db socket
-
-    Parameters
-    ----------
-    test_case : list
-        List of test_case stages (dicts with input, output and stage keys).
     """
+    Test Logic:
+        "Check that every input message in wazuh-db socket generates the adequate output to wazuh-db socket"
+    Parameters:
+        - test_case:
+            type: list
+            brief: List of test_case stages (dicts with input, output and stage keys).
+    """
+
     for index, stage in enumerate(test_case):
         if 'ignore' in stage and stage['ignore'] == "yes":
             continue
@@ -106,8 +131,22 @@ def test_wazuh_db_messages(configure_sockets_environment, connect_to_sockets_mod
             .format(index + 1, stage['stage'], expected, response)
 
 
-def test_wazuh_db_create_agent(configure_sockets_environment, connect_to_sockets_module):
-    """Check that Wazuh DB creates the agent database when a query with a new agent ID is sent"""
+def test_wazuh_db_create_agent(test_case, connect_to_sockets_module):
+    """
+    Test Logic:
+        "Check that Wazuh DB creates the agent database when a query with a new agent ID is sent.
+        Also...
+
+        But also..."
+    Checks:
+        - The received output must match with...
+        - The received output with regex must match with...
+    Parameters:
+        - test_case:
+            type: list
+            brief: List of test_case stages (dicts with input, output and stage keys).
+    """
+
     test = {"name": "Create agent",
             "description": "Wazuh DB creates automatically the agent's database the first time a query with a new agent"
                            " ID reaches it. Once the database is created, the query is processed as expected.",