sanitize filenames in the datarecipient (remove "/" etc. from the name)

Changelog

@@ -1,3 +1,5 @@
+* 2024-10-25 10.2.3
+  sanitize filenames in the datarecipient (remove "/" etc. from the name)
 * 2024-09-30 10.2.2.1
   forgot to add the recipe
 * 2024-09-30 10.2.2

coshsh/datarecipient.py

@@ -98,7 +98,7 @@ def item_write_config(self, obj, dynamic_dir, objtype, want_tool=None):
             if not want_tool or want_tool == tool:
                 for file in obj.config_files[tool]:
                     content = obj.config_files[tool][file]
-                    with io.open(os.path.join(my_target_dir, file), "w") as f:
+                    with io.open(os.path.join(my_target_dir, coshsh.util.sanitize_filename(file)), "w") as f:
                         f.write(content)
 
     def output(self, filter=None, want_tool=None):

coshsh/util.py

@@ -9,6 +9,7 @@
 import time
 import sys
 import re
+import hashlib
 import copy
 import os
 import logging
@@ -116,6 +117,17 @@ def clean_umlauts(text):
         text = text.replace(from_str, to_str)
     return text
 
+
+def sanitize_filename(filename):
+    name, ext = os.path.splitext(filename)
+    sanitized = re.sub(r'[\\/*?:"<>| ]+', '_', name)
+    if sanitized == name:
+        return filename
+    hash_suffix = hashlib.md5(filename.encode()).hexdigest()[:4]
+    sanitized_with_hash = "{}_{}{}".format(sanitized, hash_suffix, ext)
+    return sanitized_with_hash
+
+
 def setup_logging(logdir=".", logfile="coshsh.log", scrnloglevel=logging.INFO, txtloglevel=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", backup_count=2):
     logdir = os.path.abspath(logdir)
     abs_logfile = logfile if os.path.isabs(logfile) else os.path.join(logdir, logfile)

setup.py

@@ -54,7 +54,7 @@ def run(self):
 
 
 setup(name='coshsh',
-      version='10.2.2.1',
+      version='10.2.3',
       setup_requires=['wheel'],
       description='Coshsh - config generator for monitoring systems',
       long_description=open('README.md').read(),

tests/test_obdir_paths.py

@@ -0,0 +1,57 @@
+import os
+import io
+from coshsh.host import Host
+from coshsh.application import Application, GenericApplication
+from coshsh.monitoringdetail import MonitoringDetail
+from tests.common_coshsh_test import CommonCoshshTest
+
+
+# i test .../dynamic/hostgroups/...
+# hostgroup names with "/" should be stripped.
+
+
+class CoshshTest(CommonCoshshTest):
+    _configfile = 'etc/coshsh5.cfg'
+    _objectsdir = "./var/objects/test33"
+
+    def tearDown(self):
+        pass
+
+    def test_generic_app(self):
+        self.setUpConfig("etc/coshsh5.cfg", "test33")
+        r = self.generator.get_recipe("test33")
+        ds = self.generator.get_recipe("test33").get_datasource("simplesample")
+        ds.objects = r.objects
+        host = Host({
+            'host_name': 'testhost',
+            'address': '127.0.0.1',
+        })
+        ds.add('hosts', host)
+        host.hostgroups.append("servers")
+        host.hostgroups.append("08/15")
+        app = Application({
+            'host_name': 'testhost',
+            'name': 'noname',
+            'type': 'arschknarsch',
+        })
+        ds.add('applications', app)
+        r.collect()
+        r.assemble()
+        r.render()
+        self.assertTrue(len(r.objects['applications']) == 1)
+        self.assertTrue(ds.getall('applications')[0] == app)
+        self.assertTrue(ds.getall('applications')[0].__class__ == GenericApplication)
+        r.output()
+        self.assertTrue(os.path.exists("var/objects/test33/dynamic/hosts/testhost/host.cfg"))
+        self.assertTrue(os.path.exists("var/objects/test33/dynamic/hostgroups/hostgroup_servers.cfg"))
+        self.assertTrue(os.path.exists("var/objects/test33/dynamic/hostgroups/hostgroup_08_15_5641.cfg"))
+        with io.open("var/objects/test33/dynamic/hostgroups/hostgroup_servers.cfg") as f:
+            hg = f.read()
+        self.assertTrue('servers' in hg)
+        with io.open("var/objects/test33/dynamic/hostgroups/hostgroup_08_15_5641.cfg") as f:
+            hg = f.read()
+        self.assertTrue('08/15' in hg)
+        self.assertFalse('08_15' in hg)
+        self.assertTrue(not os.path.exists("var/objects/test33/dynamic/hosts/testhost/app_my_generic_fs.cfg"))
+        self.assertTrue(not os.path.exists("var/objects/test33/dynamic/hosts/testhost/app_my_generic_ports.cfg"))
+