Imp/jtt 113 update env attributes on run time

iso15118/evcc/evcc_settings.py

@@ -6,7 +6,7 @@
 import environs
 
 from iso15118.shared.network import validate_nic
-from iso15118.shared.settings import shared_settings
+from iso15118.shared.settings import load_shared_settings, shared_settings
 
 logger = logging.getLogger(__name__)
 
@@ -42,6 +42,7 @@ def load_envs(self, env_path: Optional[str] = None) -> None:
             default="iso15118/shared/examples/evcc/iso15118_2/evcc_config_eim_ac.json",
         )
         env.seal()  # raise all errors at once, if any
+        load_shared_settings()
         logger.info("EVCC environment settings:")
         for key, value in shared_settings.items():
             logger.info(f"{key:30}: {value}")

iso15118/secc/controller/simulator.py

@@ -170,7 +170,7 @@
     load_cert,
     load_priv_key,
 )
-from iso15118.shared.settings import V20_EVSE_SERVICES_CONFIG
+from iso15118.shared.settings import SettingKey, shared_settings
 from iso15118.shared.states import State
 
 logger = logging.getLogger(__name__)
@@ -273,7 +273,9 @@ def get_evse_context():
 # enum value of the energy transfer mode and value is the service parameter
 async def read_service_id_parameter_mappings():
     try:
-        async with async_open(V20_EVSE_SERVICES_CONFIG, "r") as v20_service_config:
+        async with async_open(
+            shared_settings[SettingKey.V20_SERVICE_CONFIG], "r"
+        ) as v20_service_config:
             try:
                 json_mapping = await v20_service_config.read()
                 v20_service_parameter_mapping = V20ServiceParamMapping.parse_raw(
@@ -283,11 +285,11 @@ async def read_service_id_parameter_mappings():
             except ValueError as exc:
                 raise ValueError(
                     f"Error reading 15118-20 service parameters settings file"
-                    f" at {V20_EVSE_SERVICES_CONFIG}"
+                    f" at {shared_settings[SettingKey.V20_SERVICE_CONFIG]}"
                 ) from exc
     except (FileNotFoundError, IOError) as exc:
         raise FileNotFoundError(
-            f"V20 config not found at {V20_EVSE_SERVICES_CONFIG}"
+            f"V20 config not found at {shared_settings[SettingKey.V20_SERVICE_CONFIG]}"
         ) from exc
 
 

iso15118/secc/main.py

@@ -17,7 +17,7 @@ async def main():
     """
     config = Config()
     config.load_envs()
-    config.log_settings()
+    config.print_settings()
 
     sim_evse_controller = await SimEVSEController.create()
     await sim_evse_controller.set_status(ServiceStatus.STARTING)

iso15118/secc/secc_settings.py

@@ -7,7 +7,7 @@
 
 from iso15118.secc.controller.interface import EVSEControllerInterface
 from iso15118.shared.messages.enums import AuthEnum, Protocol
-from iso15118.shared.settings import shared_settings
+from iso15118.shared.settings import load_shared_settings, shared_settings
 from iso15118.shared.utils import load_requested_auth_modes, load_requested_protocols
 
 logger = logging.getLogger(__name__)
@@ -16,7 +16,7 @@
 @dataclass
 class Config:
     iface: Optional[str] = None
-    log_level: Optional[int] = None
+    log_level: Optional[str] = None
     evse_controller: Type[EVSEControllerInterface] = None
     enforce_tls: bool = False
     free_charging_service: bool = False
@@ -37,6 +37,7 @@ class Config:
         "EIM",
         "PNC",
     ]
+    env_dump: Optional[dict] = None
 
     def load_envs(self, env_path: Optional[str] = None) -> None:
         """
@@ -102,13 +103,12 @@ def load_envs(self, env_path: Optional[str] = None) -> None:
         # enum values in PowerDeliveryReq's ChargeProgress field). In Standby, the
         # EV can still use value-added services while not consuming any power.
         self.standby_allowed = env.bool("STANDBY_ALLOWED", default=False)
-
+        load_shared_settings(env_path)
         env.seal()  # raise all errors at once, if any
-        self.secc_env = env.dump()
+        self.env_dump = dict(env.dump())
+        self.env_dump.update(shared_settings)
 
-    def log_settings(self):
+    def print_settings(self):
         logger.info("SECC settings:")
-        for key, value in shared_settings.items():
-            logger.info(f"{key:30}: {value}")
-        for key, value in self.secc_env.items():
+        for key, value in self.env_dump.items():
             logger.info(f"{key:30}: {value}")

iso15118/shared/exi_codec.py

@@ -71,7 +71,7 @@
     DCWeldingDetectionReq,
     DCWeldingDetectionRes,
 )
-from iso15118.shared.settings import MESSAGE_LOG_EXI, MESSAGE_LOG_JSON
+from iso15118.shared.settings import SettingKey, shared_settings
 
 logger = logging.getLogger(__name__)
 
@@ -244,7 +244,7 @@ def to_exi(self, msg_element: BaseModel, protocol_ns: str) -> bytes:
                                    {exc}"
             ) from exc
 
-        if MESSAGE_LOG_JSON:
+        if shared_settings[SettingKey.MESSAGE_LOG_EXI]:
             logger.info(f"Message to encode (ns={protocol_ns}): {msg_content}")
 
         try:
@@ -257,7 +257,7 @@ def to_exi(self, msg_element: BaseModel, protocol_ns: str) -> bytes:
                 f"EXIEncodingError for {str(msg_element)}: " f"{exc}"
             ) from exc
 
-        if MESSAGE_LOG_EXI:
+        if shared_settings[SettingKey.MESSAGE_LOG_EXI]:
             logger.debug(f"EXI-encoded message: {exi_stream.hex()}")
 
         return exi_stream
@@ -283,7 +283,7 @@ def from_exi(
         Raises:
             EXIDecodingError
         """
-        if MESSAGE_LOG_EXI:
+        if shared_settings[SettingKey.MESSAGE_LOG_EXI]:
             logger.debug(f"EXI-encoded message (ns={namespace}): {exi_message.hex()}")
 
         try:
@@ -300,7 +300,7 @@ def from_exi(
                 f"processing decoded EXI: {exc}"
             ) from exc
 
-        if MESSAGE_LOG_JSON:
+        if shared_settings[SettingKey.MESSAGE_LOG_JSON]:
             logger.info(f"Decoded message (ns={namespace}): {exi_decoded}")
 
         try:

iso15118/shared/security.py

@@ -80,7 +80,7 @@
     Transform,
     Transforms,
 )
-from iso15118.shared.settings import ENABLE_TLS_1_3, PKI_PATH
+from iso15118.shared.settings import SettingKey, shared_settings
 
 logger = logging.getLogger(__name__)
 
@@ -128,7 +128,7 @@ def get_ssl_context(server_side: bool) -> Optional[SSLContext]:
          as well as read the password.
     """
 
-    if ENABLE_TLS_1_3:
+    if shared_settings[SettingKey.ENABLE_TLS_1_3]:
         ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS)
     else:
         # Specifying protocol as `PROTOCOL_TLS` does best effort.
@@ -160,7 +160,7 @@ def get_ssl_context(server_side: bool) -> Optional[SSLContext]:
             logger.exception(exc)
             return None
 
-        if ENABLE_TLS_1_3:
+        if shared_settings[SettingKey.ENABLE_TLS_1_3]:
             # In 15118-20 we should also verify EVCC's certificate chain.
             # The spec however says TLS 1.3 should also support 15118-2
             # (Table 5 in V2G20 specification)
@@ -201,7 +201,7 @@ def get_ssl_context(server_side: bool) -> Optional[SSLContext]:
             "ECDHE-ECDSA-AES128-SHA256"
         )
 
-        if ENABLE_TLS_1_3:
+        if shared_settings[SettingKey.ENABLE_TLS_1_3]:
             try:
                 ssl_context.load_cert_chain(
                     certfile=CertPath.OEM_CERT_CHAIN_PEM,
@@ -1449,34 +1449,39 @@ class CertPath(str, Enum):
     """
 
     # Mobility operator (MO)
-    CONTRACT_LEAF_DER = os.path.join(PKI_PATH, "iso15118_2/certs/contractLeafCert.der")
-    MO_SUB_CA2_DER = os.path.join(PKI_PATH, "iso15118_2/certs/moSubCA2Cert.der")
-    MO_SUB_CA1_DER = os.path.join(PKI_PATH, "iso15118_2/certs/moSubCA1Cert.der")
-    MO_ROOT_DER = os.path.join(PKI_PATH, "iso15118_2/certs/moRootCACert.der")
+    CONTRACT_LEAF_DER = "contractLeafCert.der"
+    MO_SUB_CA2_DER = "moSubCA2Cert.der"
+    MO_SUB_CA1_DER = "moSubCA1Cert.der"
+    MO_ROOT_DER = "moRootCACert.der"
 
     # Charge point operator (CPO)
-    SECC_LEAF_DER = os.path.join(PKI_PATH, "iso15118_2/certs/seccLeafCert.der")
-    SECC_LEAF_PEM = os.path.join(PKI_PATH, "iso15118_2/certs/seccLeafCert.pem")
-    CPO_SUB_CA2_DER = os.path.join(PKI_PATH, "iso15118_2/certs/cpoSubCA2Cert.der")
-    CPO_SUB_CA1_DER = os.path.join(PKI_PATH, "iso15118_2/certs/cpoSubCA1Cert.der")
-    V2G_ROOT_DER = os.path.join(PKI_PATH, "iso15118_2/certs/v2gRootCACert.der")
-    V2G_ROOT_PEM = os.path.join(PKI_PATH, "iso15118_2/certs/v2gRootCACert.pem")
+    SECC_LEAF_DER = "seccLeafCert.der"
+    SECC_LEAF_PEM = "seccLeafCert.pem"
+    CPO_SUB_CA2_DER = "cpoSubCA2Cert.der"
+    CPO_SUB_CA1_DER = "cpoSubCA1Cert.der"
+    V2G_ROOT_DER = "v2gRootCACert.der"
+    V2G_ROOT_PEM = "v2gRootCACert.pem"
     # Needed for the 'certfile' parameter in ssl_context.load_cert_chain()
-    CPO_CERT_CHAIN_PEM = os.path.join(PKI_PATH, "iso15118_2/certs/cpoCertChain.pem")
+    CPO_CERT_CHAIN_PEM = "cpoCertChain.pem"
 
     # Certificate provisioning service (CPS)
-    CPS_LEAF_DER = os.path.join(PKI_PATH, "iso15118_2/certs/cpsLeafCert.der")
-    CPS_SUB_CA2_DER = os.path.join(PKI_PATH, "iso15118_2/certs/cpsSubCA2Cert.der")
-    CPS_SUB_CA1_DER = os.path.join(PKI_PATH, "iso15118_2/certs/cpsSubCA1Cert.der")
+    CPS_LEAF_DER = "cpsLeafCert.der"
+    CPS_SUB_CA2_DER = "cpsSubCA2Cert.der"
+    CPS_SUB_CA1_DER = "cpsSubCA1Cert.der"
     # The root is the V2G_ROOT
 
     # EV manufacturer (OEM)
-    OEM_LEAF_DER = os.path.join(PKI_PATH, "iso15118_2/certs/oemLeafCert.der")
-    OEM_SUB_CA2_DER = os.path.join(PKI_PATH, "iso15118_2/certs/oemSubCA2Cert.der")
-    OEM_SUB_CA1_DER = os.path.join(PKI_PATH, "iso15118_2/certs/oemSubCA1Cert.der")
-    OEM_ROOT_DER = os.path.join(PKI_PATH, "iso15118_2/certs/oemRootCACert.der")
-    OEM_ROOT_PEM = os.path.join(PKI_PATH, "iso15118_2/certs/oemRootCACert.pem")
-    OEM_CERT_CHAIN_PEM = os.path.join(PKI_PATH, "iso15118_2/certs/oemCertChain.pem")
+    OEM_LEAF_DER = "oemLeafCert.der"
+    OEM_SUB_CA2_DER = "oemSubCA2Cert.der"
+    OEM_SUB_CA1_DER = "oemSubCA1Cert.der"
+    OEM_ROOT_DER = "oemRootCACert.der"
+    OEM_ROOT_PEM = "oemRootCACert.pem"
+    OEM_CERT_CHAIN_PEM = "oemCertChain.pem"
+
+    def __get__(self, instance, owner):
+        return os.path.join(
+            shared_settings[SettingKey.PKI_PATH], "iso15118_2/certs/", self.value
+        )
 
 
 class KeyPath(str, Enum):
@@ -1489,30 +1494,35 @@ class KeyPath(str, Enum):
     """
 
     # Mobility operator (MO)
-    CONTRACT_LEAF_PEM = os.path.join(
-        PKI_PATH, "iso15118_2/private_keys/contractLeaf" ".key"
-    )
-    MO_SUB_CA2_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/moSubCA2.key")
-    MO_SUB_CA1_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/moSubCA1.key")
-    MO_ROOT_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/moRootCA.key")
+    CONTRACT_LEAF_PEM = "contractLeaf.key"
+    MO_SUB_CA2_PEM = "moSubCA2.key"
+    MO_SUB_CA1_PEM = "moSubCA1.key"
+    MO_ROOT_PEM = "moRootCA.key"
 
     # Charge point operator (CPO)
-    SECC_LEAF_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/seccLeaf.key")
-    CPO_SUB_CA2_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/cpoSubCA2.key")
-    CPO_SUB_CA1_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/cpoSubCA1.key")
-    V2G_ROOT_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/v2gRootCA.key")
+    SECC_LEAF_PEM = "seccLeaf.key"
+    CPO_SUB_CA2_PEM = "cpoSubCA2.key"
+    CPO_SUB_CA1_PEM = "cpoSubCA1.key"
+    V2G_ROOT_PEM = "v2gRootCA.key"
 
     # Certificate provisioning service (CPS)
-    CPS_LEAF_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/cpsLeaf.key")
-    CPS_SUB_CA2_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/cpsSubCA2.key")
-    CPS_SUB_CA1_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/cpsSubCA1.key")
+    CPS_LEAF_PEM = "cpsLeaf.key"
+    CPS_SUB_CA2_PEM = "cpsSubCA2.key"
+    CPS_SUB_CA1_PEM = "cpsSubCA1.key"
     # The root is the V2G_ROOT
 
     # EV manufacturer (OEM)
-    OEM_LEAF_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/oemLeaf.key")
-    OEM_SUB_CA2_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/oemSubCA2.key")
-    OEM_SUB_CA1_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/oemSubCA1.key")
-    OEM_ROOT_PEM = os.path.join(PKI_PATH, "iso15118_2/private_keys/oemRootCA.key")
+    OEM_LEAF_PEM = "oemLeaf.key"
+    OEM_SUB_CA2_PEM = "oemSubCA2.key"
+    OEM_SUB_CA1_PEM = "oemSubCA1.key"
+    OEM_ROOT_PEM = "oemRootCA.key"
+
+    def __get__(self, instance, owner):
+        return os.path.join(
+            shared_settings[SettingKey.PKI_PATH],
+            "iso15118_2/private_keys/",
+            self.value,
+        )
 
 
 class KeyPasswordPath(str, Enum):
@@ -1524,18 +1534,15 @@ class KeyPasswordPath(str, Enum):
     """
 
     # Private key password paths
-    SECC_LEAF_KEY_PASSWORD = os.path.join(
-        PKI_PATH, "iso15118_2/private_keys/seccLeafPassword.txt"
-    )
-    OEM_LEAF_KEY_PASSWORD = os.path.join(
-        PKI_PATH, "iso15118_2/private_keys/oemLeafPassword.txt"
-    )
-    CONTRACT_LEAF_KEY_PASSWORD = os.path.join(
-        PKI_PATH, "iso15118_2/private_keys/contractLeafPassword.txt"
-    )
-    CPS_LEAF_KEY_PASSWORD = os.path.join(
-        PKI_PATH, "iso15118_2/private_keys/cpsLeafPassword.txt"
-    )
-    MO_SUB_CA2_PASSWORD = os.path.join(
-        PKI_PATH, "iso15118_2/private_keys/moSubCA2LeafPassword.txt"
-    )
+    SECC_LEAF_KEY_PASSWORD = "seccLeafPassword.txt"
+    OEM_LEAF_KEY_PASSWORD = "oemLeafPassword.txt"
+    CONTRACT_LEAF_KEY_PASSWORD = "contractLeafPassword.txt"
+    CPS_LEAF_KEY_PASSWORD = "cpsLeafPassword.txt"
+    MO_SUB_CA2_PASSWORD = "moSubCA2LeafPassword.txt"
+
+    def __get__(self, instance, owner):
+        return os.path.join(
+            shared_settings[SettingKey.PKI_PATH],
+            "iso15118_2/private_keys/",
+            self.value,
+        )

iso15118/shared/settings.py

@@ -1,28 +1,37 @@
 import os
+from typing import Optional
 
 import environs
 
-SHARED_CWD = os.path.dirname(os.path.abspath(__file__))
-JAR_FILE_PATH = SHARED_CWD + "/EXICodec.jar"
-
-WORK_DIR = os.getcwd()
 
-ENV_PATH = WORK_DIR + "/.env"
+class SettingKey:
+    PKI_PATH = "PKI_PATH"
+    MESSAGE_LOG_JSON = "MESSAGE_LOG_JSON"
+    MESSAGE_LOG_EXI = "MESSAGE_LOG_EXI"
+    V20_SERVICE_CONFIG = "V20_SERVICE_CONFIG"
+    ENABLE_TLS_1_3 = "ENABLE_TLS_1_3"
 
-env = environs.Env(eager=False)
-env.read_env(path=ENV_PATH)  # read .env file, if it exists
 
-PKI_PATH = env.path("PKI_PATH", default=SHARED_CWD + "/pki/")
-
-MESSAGE_LOG_JSON = env.bool("MESSAGE_LOG_JSON", default=True)
-MESSAGE_LOG_EXI = env.bool("MESSAGE_LOG_EXI", default=False)
+shared_settings = {}
+SHARED_CWD = os.path.dirname(os.path.abspath(__file__))
+JAR_FILE_PATH = SHARED_CWD + "/EXICodec.jar"
 
-V20_EVSE_SERVICES_CONFIG = env.str(
-    "V20_SERVICE_CONFIG",
-    default=SHARED_CWD + "/examples/secc/15118_20/service_config.json",
-)
+WORK_DIR = os.getcwd()
 
-ENABLE_TLS_1_3 = env.bool("ENABLE_TLS_1_3", default=False)
 
-env.seal()  # raise all errors at once, if any
-shared_settings = env.dump()
+def load_shared_settings(env_path: Optional[str] = None):
+    env = environs.Env(eager=False)
+    env.read_env(path=env_path)  # read .env file, if it exists
+
+    settings = {
+        SettingKey.PKI_PATH: env.str("PKI_PATH", default=SHARED_CWD + "/pki/"),
+        SettingKey.MESSAGE_LOG_JSON: env.bool("MESSAGE_LOG_JSON", default=True),
+        SettingKey.MESSAGE_LOG_EXI: env.bool("MESSAGE_LOG_EXI", default=False),
+        SettingKey.V20_SERVICE_CONFIG: env.str(
+            "V20_SERVICE_CONFIG",
+            default=SHARED_CWD + "/examples/secc/15118_20/service_config.json",
+        ),
+        SettingKey.ENABLE_TLS_1_3: env.bool("ENABLE_TLS_1_3", default=False),
+    }
+    shared_settings.update(settings)
+    env.seal()  # raise all errors at once, if any