Move assign_wazuh_ownership() to utils

api/scripts/wazuh_apid.py

@@ -49,22 +49,6 @@ def spawn_authentication_pool():
     signal.signal(signal.SIGINT, signal.SIG_IGN)
 
 
-def assign_wazuh_ownership(filepath: str):
-    """Create a file if it doesn't exist and assign ownership.
-
-    Parameters
-    ----------
-    filepath : str
-        File to assign ownership.
-    """
-    if not os.path.isfile(filepath):
-        f = open(filepath, "w")
-        f.close()
-    if os.stat(filepath).st_gid != common.wazuh_gid() or \
-        os.stat(filepath).st_uid != common.wazuh_uid():
-        os.chown(filepath, common.wazuh_uid(), common.wazuh_gid())
-
-
 def configure_ssl(params):
     """Configure https files and permission, and set the uvicorn dictionary configuration keys.
 
@@ -106,8 +90,8 @@ def configure_ssl(params):
                 logger.warning(SSL_DEPRECATED_MESSAGE.format(ssl_protocol=config_ssl_protocol))
 
         # Check and assign ownership to wazuh user for server.key and server.crt files
-        assign_wazuh_ownership(api_conf['https']['key'])
-        assign_wazuh_ownership(api_conf['https']['cert'])
+        utils.assign_wazuh_ownership(api_conf['https']['key'])
+        utils.assign_wazuh_ownership(api_conf['https']['cert'])
 
         params['ssl_version'] = ssl.PROTOCOL_TLS_SERVER
 
@@ -386,7 +370,7 @@ def error(self, msg, *args, **kws):
     # set permission on log files
     for handler in uvicorn_params['log_config']['handlers'].values():
         if 'filename' in handler:
-            assign_wazuh_ownership(handler['filename'])
+            utils.assign_wazuh_ownership(handler['filename'])
             os.chmod(handler['filename'], 0o660)
 
     # Configure and create the wazuh-api logger

framework/wazuh/core/tests/test_utils.py

@@ -160,6 +160,34 @@ def to_dict(self):
 <!-- Example -->
 '''
 
+@patch('os.chown')
+@patch('wazuh.core.common.wazuh_uid')
+@patch('wazuh.core.common.wazuh_gid')
+def test_assign_wazuh_ownership(mock_gid, mock_uid, mock_chown):
+    """Test assign_wazuh_ownership function."""
+    with TemporaryDirectory() as tmp_dirname:
+        tmp_file = NamedTemporaryFile(dir=tmp_dirname, delete=False)
+        filename = os.path.join(tmp_dirname, tmp_file.name)
+        utils.assign_wazuh_ownership(filename)
+
+        mock_chown.assert_called_once_with(filename, mock_uid(), mock_gid())
+
+
+@patch('os.chown')
+@patch('wazuh.core.common.wazuh_uid')
+@patch('wazuh.core.common.wazuh_gid')
+def test_assign_wazuh_ownership_write_file(mock_gid, mock_uid, mock_chown):
+    """Test assign_wazuh_ownership function with a non-regular file."""
+    with TemporaryDirectory() as tmp_dirname:
+        tmp_file = NamedTemporaryFile(dir=tmp_dirname, delete=False)
+        filename = os.path.join(tmp_dirname, tmp_file.name)
+
+        with patch('os.path.isfile', return_value=False):
+            with patch('builtins.open') as mock_open:
+                utils.assign_wazuh_ownership(filename)
+                mock_open.assert_called_once_with(filename, 'w')
+
+            mock_chown.assert_called_once_with(filename, mock_uid(), mock_gid())
 
 @pytest.mark.parametrize('month', [
     1,

framework/wazuh/core/utils.py

@@ -42,6 +42,21 @@
 t_cache = TTLCache(maxsize=4500, ttl=60)
 
 
+def assign_wazuh_ownership(filepath: str):
+    """Create a file if it doesn't exist and assign ownership.
+
+    Parameters
+    ----------
+    filepath : str
+        File to assign ownership.
+    """
+    if not os.path.isfile(filepath):
+        f = open(filepath, "w")
+        f.close()
+    if os.stat(filepath).st_gid != common.wazuh_gid() or \
+        os.stat(filepath).st_uid != common.wazuh_uid():
+        os.chown(filepath, common.wazuh_uid(), common.wazuh_gid())
+
 def clean_pid_files(daemon: str) -> None:
     """Check the existence of '.pid' files for a specified daemon.