add security and persistent store adaptions

Signed-off-by: Fabian Klemm <[email protected]>

everest-testing/README.md

@@ -27,13 +27,17 @@ The core_utils basically provide two fixtures that you can require in your test
 
 - **core_config** Core configuration, which is the everest_core path and the configuration path (utilizes the `everest_core_config` marker.)
 - **probe_module_config** The main fixture `everest_core` can be used to start and stop the everest-core application.
-- **ocpp_config** The main fixture `everest_core` can be used to start and stop the everest-core application.
+- **ocpp_config**  Used to provide the configuration to set up the OCPP (1.6 or 2.0.1) module.
+- **evse_security_config** Used to provide the configuration to set up the EvseSecurity module.
+- **persistent_storage_config** Used to provide the configuration to set up the PersistentStorage module.
 
 ### pytest markers
 
 Some OCPP fixtures will parse pytest markers of test cases. The following markers can be used:
 - **everest_core_config**: Can be used to specify the everest configuration file to be used in this test case
 - **probe_module**: If set, the ProbeModule will be injected into the config (used by the `probe_module_config` fixture). You may provide the config args as fixture arguments.
+- **source_certs_dir**: If set and the  default `evse_security_config` fixture is used, this will cause the  `EvseSecurity` module configuration to use a  temporary certificates folder into which the source certificate folder trees are copied.
+- **use_temporary_persistent_store**: If set and the  default `persistent_storage_config` fixture is used, this will cause the  `PersistentStore` module configuration to use a  temporary database.
 
 ## OCPP utils
 
@@ -77,7 +81,7 @@ Note: When overriding a fixture, be careful which pytest markers might be ignore
 
 ```python
 
-from everest.testing.core_utils.fixtures import everest_core, probe_module_config, test_controller
+from everest.testing.core_utils.fixtures import *
 from everest.testing.ocpp_utils.fixtures import ocpp_config
 
 @pytest.fixture
@@ -103,6 +107,8 @@ class TestMyEverestModule:
 
 ```
 
+_Note_: The "*" import from `core_utils.fixtures` may ensure backwards compatibility to automatically load new default fixtures in the future!
+
 
 ## Install
 

everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/evse_security_configuration_visitor.py

@@ -0,0 +1,78 @@
+from copy import deepcopy
+from dataclasses import dataclass, asdict
+from pathlib import Path
+from typing import Dict, Optional
+
+from everest.testing.core_utils.configuration.everest_configuration_visitors.everst_configuration_visitor import \
+    EverestConfigAdjustmentVisitor
+
+
+@dataclass
+class EvseSecurityModuleConfiguration:
+    csms_ca_bundle: Optional[str] = None
+    mf_ca_bundle: Optional[str] = None
+    mo_ca_bundle: Optional[str] = None
+    v2g_ca_bundle: Optional[str] = None
+    csms_leaf_cert_directory: Optional[str] = None
+    csms_leaf_key_directory: Optional[str] = None
+    secc_leaf_cert_directory: Optional[str] = None
+    secc_leaf_key_directory: Optional[str] = None
+    private_key_password: Optional[str] = None
+
+    @staticmethod
+    def default_from_cert_path(certs_directory: Path):
+        """ Return a default definition of bundles and directory given base directory. """
+        return EvseSecurityModuleConfiguration(
+            csms_ca_bundle=str(certs_directory / "ca/csms/CSMS_ROOT_CA.pem"),
+            mf_ca_bundle=str(certs_directory / "ca/mf/MF_ROOT_CA.pem"),
+            mo_ca_bundle=str(certs_directory / "ca/mo/MO_ROOT_CA.pem"),
+            v2g_ca_bundle=str(certs_directory / "ca/v2g/V2G_ROOT_CA.pem"),
+            csms_leaf_cert_directory=str(certs_directory / "client/csms"),
+            csms_leaf_key_directory=str(certs_directory / "client/csms"),
+            secc_leaf_cert_directory=str(certs_directory / "client/cso"),
+            secc_leaf_key_directory=str(certs_directory / "client/cso"),
+        )
+
+    def to_module_configuration(self) -> Dict:
+        return {k: v for k, v in asdict(self).items() if v is not None}
+
+
+class EvseSecurityModuleConfigurationVisitor(EverestConfigAdjustmentVisitor):
+    """ Adjusts the Everest configuration by manipulating the evse security module configuration to point
+    to the correct (temporary) certificate paths.
+
+    """
+
+    def __init__(self,
+                 configuration: EvseSecurityModuleConfiguration,
+                 module_id: Optional[str] = None):
+        """
+
+        Args:
+            configuration: module configuration. This will be merged into the template configuration (meaning None values are ignored/taken from the originally provided Everest configuration)
+            module_id: Id of security module; if None, auto-detected by module type "EvseSecurity
+        """
+        self._security_module_id = module_id
+        self._configuration = configuration
+
+    def _determine_module_id(self, everest_config: Dict):
+        if self._security_module_id:
+            assert self._security_module_id in everest_config[
+                "active_modules"], f"Module id {self._security_module_id} not found in EVerest configuration"
+            return self._security_module_id
+        else:
+            try:
+                return next(k for k, v in everest_config["active_modules"].items() if v["module"] == "EvseSecurity")
+            except StopIteration:
+                raise ValueError("No EvseSecurity module found in EVerest configuration")
+
+    def adjust_everest_configuration(self, everest_config: Dict):
+
+        adjusted_config = deepcopy(everest_config)
+
+        module_cfg = adjusted_config["active_modules"][self._determine_module_id(adjusted_config)]
+
+        module_cfg["config_module"] = {**module_cfg["config_module"],
+                                       **self._configuration.to_module_configuration()}
+
+        return adjusted_config

everest-testing/src/everest/testing/core_utils/configuration/everest_configuration_visitors/persistent_store_configuration_visitor.py

@@ -0,0 +1,47 @@
+from copy import deepcopy
+from dataclasses import dataclass, asdict
+from pathlib import Path
+from typing import Dict, Optional
+
+from everest.testing.core_utils.configuration.everest_configuration_visitors.everst_configuration_visitor import \
+    EverestConfigAdjustmentVisitor
+
+
+class PersistentStoreConfigurationVisitor(EverestConfigAdjustmentVisitor):
+    """ Adjusts the Everest configuration by manipulating the PersistentStore module configuration to point
+    to the desired (temporary) storage
+
+    """
+
+    def __init__(self,
+                 sqlite_db_file_path: Path,
+                 module_id: Optional[str] = None):
+        """
+
+        Args:
+            sqlite_db_file_path: database to be used in configuration
+            module_id: Id of security module; if None, auto-detected by module type "EvseSecurity
+        """
+        self._module_id = module_id
+        self._sqlite_db_file_path = sqlite_db_file_path
+
+    def _determine_module_id(self, everest_config: Dict):
+        if self._module_id:
+            assert self._module_id in everest_config[
+                "active_modules"], f"Module id {self._module_id} not found in EVerest configuration"
+            return self._module_id
+        else:
+            try:
+                return next(k for k, v in everest_config["active_modules"].items() if v["module"] == "PersistentStore")
+            except StopIteration:
+                raise ValueError("No PersistentStore module found in EVerest configuration")
+
+    def adjust_everest_configuration(self, everest_config: Dict):
+
+        adjusted_config = deepcopy(everest_config)
+
+        module_cfg = adjusted_config["active_modules"][self._determine_module_id(adjusted_config)]
+
+        module_cfg.setdefault("config_module", {})["sqlite_db_file_path"] = str(self._sqlite_db_file_path)
+
+        return adjusted_config

everest-testing/src/everest/testing/core_utils/configuration/everest_environment_setup.py

@@ -2,15 +2,21 @@
 # Copyright 2020 - 2023 Pionix GmbH and Contributors to EVerest
 
 import logging
+import shutil
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Optional, Dict, List, Union
 
 import yaml
 
 from everest.testing.core_utils.common import OCPPVersion
-from everest.testing.core_utils.configuration.everest_configuration_visitors.ocpp_module_configuration_visitor import OCPPModuleConfigurationVisitor, \
+from everest.testing.core_utils.configuration.everest_configuration_visitors.evse_security_configuration_visitor import \
+    EvseSecurityModuleConfigurationVisitor, EvseSecurityModuleConfiguration
+from everest.testing.core_utils.configuration.everest_configuration_visitors.ocpp_module_configuration_visitor import \
+    OCPPModuleConfigurationVisitor, \
     OCPPModulePaths16, OCPPModulePaths201
+from everest.testing.core_utils.configuration.everest_configuration_visitors.persistent_store_configuration_visitor import \
+    PersistentStoreConfigurationVisitor
 from everest.testing.core_utils.configuration.everest_configuration_visitors.probe_module_configuration_visitor import \
     ProbeModuleConfigurationVisitor
 
@@ -31,6 +37,18 @@ class EverestEnvironmentOCPPConfiguration:
         Path] = None  # Path for OCPP config to be used; if not provided, will be determined from everest config
 
 
+@dataclass
+class EverestEnvironmentEvseSecurityConfiguration:
+    use_temporary_certificates_folder: bool = True  # if true, configuration will be adapted to use temporary certifcates folder, this assumes a "default" file tree structure, cf. the EvseSecurityModuleConfiguration dataclass
+    module_id: Optional[str] = None  # if None, auto-detected
+    source_certificate_directory: Optional[Path] = None  # if provided, this will be copied to temporary path and
+
+
+@dataclass
+class EverestEnvironmentPersistentStoreConfiguration:
+    use_temporary_folder: bool = True  # if true, a temporary persistant storage folder will be used
+
+
 @dataclass
 class EverestEnvironmentCoreConfiguration:
     everest_core_path: Path
@@ -62,21 +80,26 @@ class EverestTestEnvironmentSetup:
     @dataclass
     class _EverestEnvironmentTemporaryPaths:
         """ Paths of the temporary configuration files / data """
+        certs_dir: Path  # used by both OCPP and evse security
         ocpp_config_file: Path
         ocpp_user_config_file: Path
         ocpp_database_dir: Path
-        ocpp_certs_dir: Path
         ocpp_message_log_directory: Path
+        persistent_store_db_path: Path
 
     def __init__(self,
                  core_config: EverestEnvironmentCoreConfiguration,
                  ocpp_config: Optional[EverestEnvironmentOCPPConfiguration] = None,
                  probe_config: Optional[EverestEnvironmentProbeModuleConfiguration] = None,
+                 evse_security_config: Optional[EverestEnvironmentEvseSecurityConfiguration] = None,
+                 persistent_store_config: Optional[EverestEnvironmentPersistentStoreConfiguration] = None,
                  standalone_module: Optional[str] = None
                  ) -> None:
         self._core_config = core_config
         self._ocpp_config = ocpp_config
         self._probe_config = probe_config
+        self._evse_security_config = evse_security_config
+        self._persistent_store_config = persistent_store_config
         self._standalone_module = standalone_module
         if not self._standalone_module and self._probe_config:
             self._standalone_module = self._probe_config.module_id
@@ -99,6 +122,9 @@ def setup_environment(self, tmp_path: Path):
                 temporary_paths=temporary_paths
             )
 
+        if self._evse_security_config:
+            self._setup_evse_security_configuration(temporary_paths)
+
     @property
     def everest_core(self) -> EverestCore:
         assert self._everest_core, "Everest Core not initialized; run 'setup_environment' first"
@@ -107,19 +133,23 @@ def everest_core(self) -> EverestCore:
     def _create_temporary_directory_structure(self, tmp_path: Path) -> _EverestEnvironmentTemporaryPaths:
         ocpp_config_dir = tmp_path / "ocpp_config"
         ocpp_config_dir.mkdir(exist_ok=True)
-        ocpp_certs_dir = ocpp_config_dir / "certs"
-        ocpp_certs_dir.mkdir(exist_ok=True)
+        certs_dir = tmp_path / "certs"
+        certs_dir.mkdir(exist_ok=True)
         ocpp_logs_dir = ocpp_config_dir / "logs"
         ocpp_logs_dir.mkdir(exist_ok=True)
 
+        persistent_store_dir = tmp_path / "persistent_storage"
+        persistent_store_dir.mkdir(exist_ok=True)
+
         logging.info(f"temp ocpp config files directory: {ocpp_config_dir}")
 
         return self._EverestEnvironmentTemporaryPaths(
             ocpp_config_file=ocpp_config_dir / "config.json",
             ocpp_user_config_file=ocpp_config_dir / "user_config.json",
             ocpp_database_dir=ocpp_config_dir,
-            ocpp_certs_dir=ocpp_certs_dir,
-            ocpp_message_log_directory=ocpp_logs_dir
+            certs_dir=certs_dir,
+            ocpp_message_log_directory=ocpp_logs_dir,
+            persistent_store_db_path=persistent_store_dir / "persistent_store.db"
         )
 
     def _create_ocpp_module_configuration_visitor(self,
@@ -129,15 +159,15 @@ def _create_ocpp_module_configuration_visitor(self,
             ocpp_paths = OCPPModulePaths16(
                 ChargePointConfigPath=str(temporary_paths.ocpp_config_file),
                 MessageLogPath=str(temporary_paths.ocpp_message_log_directory),
-                CertsPath=str(temporary_paths.ocpp_certs_dir),
+                CertsPath=str(temporary_paths.certs_dir),
                 UserConfigPath=str(temporary_paths.ocpp_user_config_file),
                 DatabasePath=str(temporary_paths.ocpp_database_dir)  # self.temp_ocpp_database_dir.name
             )
         elif self._ocpp_config.ocpp_version == OCPPVersion.ocpp201:
             ocpp_paths = OCPPModulePaths201(
                 ChargePointConfigPath=str(temporary_paths.ocpp_config_file),
                 MessageLogPath=str(temporary_paths.ocpp_message_log_directory),
-                CertsPath=str(temporary_paths.ocpp_certs_dir),
+                CertsPath=str(temporary_paths.certs_dir),
                 CoreDatabasePath=str(temporary_paths.ocpp_database_dir),
                 DeviceModelDatabasePath=str(temporary_paths.ocpp_database_dir / "device_model_storage.db"),
             )
@@ -162,7 +192,7 @@ def _setup_libocpp_configuration(self, source_certs_directory: Path,
 
         liboccp_configuration_helper.install_default_ocpp_certificates(
             source_certs_directory=source_certs_directory,
-            target_certs_directory=temporary_paths.ocpp_certs_dir)  # Path(self.temp_ocpp_certs_dir.name))
+            target_certs_directory=temporary_paths.certs_dir)  # Path(self.temp_ocpp_certs_dir.name))
 
         liboccp_configuration_helper.generate_ocpp_config(
             central_system_port=self._ocpp_config.central_system_port,
@@ -187,6 +217,18 @@ def _create_everest_configuration_visitors(self, temporary_paths: _EverestEnviro
             configuration_visitors.append(
                 ProbeModuleConfigurationVisitor(connections=self._probe_config.connections,
                                                 module_id=self._probe_config.module_id))
+
+        if self._evse_security_config and self._evse_security_config.use_temporary_certificates_folder:
+            configuration_visitors.append(
+                EvseSecurityModuleConfigurationVisitor(module_id=self._evse_security_config.module_id,
+                                                       configuration=EvseSecurityModuleConfiguration.default_from_cert_path(
+                                                           temporary_paths.certs_dir)))
+
+        if self._persistent_store_config and self._persistent_store_config.use_temporary_folder:
+            configuration_visitors.append(
+                PersistentStoreConfigurationVisitor(sqlite_db_file_path=temporary_paths.persistent_store_db_path)
+            )
+
         return configuration_visitors
 
     def _determine_configured_charge_point_config_path_from_everest_config(self):
@@ -205,3 +247,9 @@ def _determine_configured_charge_point_config_path_from_everest_config(self):
             raise ValueError(f"unknown OCPP version {self._ocpp_config.ocpp_version}")
         ocpp_config_path = ocpp_dir / charge_point_config_path
         return ocpp_config_path
+
+    def _setup_evse_security_configuration(self, temporary_paths: _EverestEnvironmentTemporaryPaths):
+        """ If configures, copies the source certificate trees"""
+        if self._evse_security_config and self._evse_security_config.source_certificate_directory:
+            shutil.copytree(self._evse_security_config.source_certificate_directory / "ca", temporary_paths.certs_dir / "ca", dirs_exist_ok=True)
+            shutil.copytree(self._evse_security_config.source_certificate_directory / "client", temporary_paths.certs_dir / "client", dirs_exist_ok=True)

everest-testing/src/everest/testing/core_utils/fixtures.py

@@ -5,8 +5,10 @@
 
 import pytest
 
-from everest.testing.core_utils.configuration.everest_environment_setup import EverestEnvironmentProbeModuleConfiguration, \
-    EverestTestEnvironmentSetup, EverestEnvironmentOCPPConfiguration, EverestEnvironmentCoreConfiguration
+from everest.testing.core_utils.configuration.everest_environment_setup import \
+    EverestEnvironmentProbeModuleConfiguration, \
+    EverestTestEnvironmentSetup, EverestEnvironmentOCPPConfiguration, EverestEnvironmentCoreConfiguration, \
+    EverestEnvironmentEvseSecurityConfiguration, EverestEnvironmentPersistentStoreConfiguration
 from everest.testing.core_utils.controller.everest_test_controller import EverestTestController
 from everest.testing.core_utils.everest_core import EverestCore
 
@@ -43,13 +45,31 @@ def core_config(request) -> EverestEnvironmentCoreConfiguration:
 def ocpp_config(request) -> Optional[EverestEnvironmentOCPPConfiguration]:
     return None
 
+@pytest.fixture
+def evse_security_config(request) -> Optional[EverestEnvironmentEvseSecurityConfiguration]:
+    source_certs_dir_marker = request.node.get_closest_marker("source_certs_dir")
+    if source_certs_dir_marker:
+        return EverestEnvironmentEvseSecurityConfiguration(source_certificate_directory=Path(source_certs_dir_marker.args[0]))
+    return None
+
+@pytest.fixture
+def persistent_store_config(request) -> Optional[EverestEnvironmentPersistentStoreConfiguration]:
+    persistent_store_marker = request.node.get_closest_marker("use_temporary_persistent_store")
+    if persistent_store_marker:
+        return EverestEnvironmentPersistentStoreConfiguration(use_temporary_folder=True)
+    return None
+
+
 
 @pytest.fixture
 def everest_core(request,
                  tmp_path,
                  core_config: EverestEnvironmentCoreConfiguration,
-                 ocpp_config: EverestEnvironmentOCPPConfiguration,
-                 probe_module_config: EverestEnvironmentProbeModuleConfiguration) -> EverestCore:
+                 ocpp_config: Optional[EverestEnvironmentOCPPConfiguration],
+                 probe_module_config: Optional[EverestEnvironmentProbeModuleConfiguration],
+                 evse_security_config: Optional[EverestEnvironmentEvseSecurityConfiguration],
+                 persistent_store_config: Optional[EverestEnvironmentPersistentStoreConfiguration],
+                 ) -> EverestCore:
     """Fixture that can be used to start and stop everest-core"""
 
     standalone_module_marker = request.node.get_closest_marker('standalone_module')
@@ -58,6 +78,8 @@ def everest_core(request,
         core_config=core_config,
         ocpp_config=ocpp_config,
         probe_config=probe_module_config,
+        evse_security_config=evse_security_config,
+        persistent_store_config=persistent_store_config,
         standalone_module=standalone_module_marker
     )