add functionality for custom pkl files and zipped archives

osp/models/catalytic/co_catalyticfoam.py

@@ -251,6 +251,9 @@ def __post_init_post_parse__(self):
             if self.species_from_upload:
                 pkl = emmo.PKLFile(uid=self.species_from_upload)
                 calc.add(pkl, rel=emmo.hasCalculationInput)
+            if self.patches_from_upload:
+                tar = emmo.TarballFile(uid=self.patches_from_upload)
+                calc.add(tar, rel=emmo.hasCalculationInput)
         file = tempfile.NamedTemporaryFile(suffix=".ttl")
         export_cuds(session, file.name)
         self._uuid = get_upload(file)

osp/wrappers/simcatalyticfoam/catalyticfoam_engine.py

@@ -2,6 +2,7 @@
 import os
 import shutil
 import subprocess  # nosec
+import tarfile
 import tempfile
 from typing import TYPE_CHECKING, Optional
 
@@ -38,6 +39,7 @@ def __init__(self, case: str, config: dict = {}):
                 filename=tar,
             )
         self._pkl: Optional[str] = None
+        self._input_tarball: Optional[str] = None
         self._parse_files = dict()
         self._config: dict = config
         self._current_process: subprocess.Popen = None
@@ -229,6 +231,41 @@ def exit_code(cls):
     def tarball(cls):
         return f"{cls._config['filename']}.tar"
 
+    @property
+    def input_tarball(cls):
+        return cls._input_tarball
+
+    @input_tarball.setter
+    def _set_tarball(self, value):
+        self._input_tarball = value
+        logger.info("Received new value for case directory to be used: `%s`.", value)
+        if "gz" in value:
+            mode = "r:gz"
+        elif "bz2" in value:
+            mode = "r:bz2"
+        else:
+            mode = "r"
+        logger.info(
+            "Will remove previous files in runtime directory: `%s`",
+            self._config["directory"],
+        )
+        shutil.rmtree(self._config["directory"])
+        logger.info(
+            "Will untar provided file in to runtime directory: `%s`",
+            self._config["directory"],
+        )
+        allowed_members = settings.allowed_members.strip().split(";")
+        with tarfile.open(value, mode=mode) as file:
+            # Iterate over the members and validate
+            for member in file.getmembers():
+                if member.name in allowed_members:
+                    file.extract(
+                        member, path=self._config["directory"], set_attrs=False
+                    )
+                else:
+                    logger.warning("Skipping member: %s", member.name)
+        logger.info("Untaring complete.")
+
     @property
     def pkl(cls):
         return cls._pkl

osp/wrappers/simcatalyticfoam/settings.py

@@ -53,6 +53,14 @@ class CatalyticFoamSettings(BaseSettings):
         E.g. `fixedValue` for a Dirichlet boundary condition.""",
     )
 
+    allowed_members: str = Field(
+        "0;constant;kinetic;system;CatalyticReactors;ml_ExtraTrees_forCFD.pkl",
+        description="""If an tar-file is used as custom use case directory, the
+                                  `;`-seprated values in the provided string are the only members which
+                                  are allowed to be extracted from the tarball-file (due to security reasons,
+                                  see [B202:tarfile_unsafe_members] for reference.)""",
+    )
+
 
 class ReaxProSettings(ModelSettings, CatalyticFoamSettings):
     """General Reaxpro wrapper settings"""

osp/wrappers/simcatalyticfoam/simcatalyticfoam.py

@@ -35,6 +35,9 @@ def _run(self, root_object) -> None:
 
     # OVERRIDE
     def _apply_added(self, root_object, buffer) -> None:
+        for obj in buffer.values():
+            if obj.is_a(emmo.TarballFile):
+                self._engine.input_tarball = get_download(str(obj.uid), as_file=True)
         for obj in buffer.values():
             self._wrap(obj)