Check central path is writeable during SSH setup (#449)

datashuttle/datashuttle_class.py

@@ -1457,14 +1457,28 @@ def _make_project_metadata_if_does_not_exist(self) -> None:
         """
         folders.create_folders(self.cfg.project_metadata_path, log=False)
 
-    def _setup_rclone_central_ssh_config(self, log: bool) -> None:
+    def _setup_rclone_central_ssh_config(self, log: bool) -> None:  # 449
+        """
+        1. Set up RClone SSH config.
+        2. Check write permissions on central_path using the same SSH connection.
+        """
         rclone.setup_rclone_config_for_ssh(
             self.cfg,
             self.cfg.get_rclone_config_name("ssh"),
             self.cfg.ssh_key_path,
             log=log,
         )
 
+        if not ssh.check_write_permissions(
+            self.cfg, self.cfg["central_path"], log=log
+        ):
+            raise PermissionError(
+                f"Cannot write to central path: {self.cfg['central_path']}\n"
+            )
+
+        if log:
+            utils.log("SSH and RClone setup completed successfully.")
+
     def _setup_rclone_central_local_filesystem_config(self) -> None:
         rclone.setup_rclone_config_for_local_filesystem(
             self.cfg.get_rclone_config_name("local_filesystem"),

datashuttle/utils/ssh.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -329,3 +330,62 @@ def get_list_of_folder_names_over_sftp(
             utils.log_and_message(f"No file found at {search_path.as_posix()}")
 
     return all_folder_names, all_filenames
+
+
+# 449
+@dataclass
+class SSHCommandResult:
+    stdout: str
+    stderr: str
+    returncode: int
+
+
+def run_ssh_command(
+    cfg: Configs,
+    command: str,
+    password: Optional[str] = None,
+    log: bool = True,
+) -> SSHCommandResult:
+    """
+    Run a command on the remote server over SSH and return a result object.
+    """
+    client = paramiko.SSHClient()
+    try:
+        connect_client_with_logging(
+            client, cfg, password, message_on_sucessful_connection=log
+        )
+        stdin, stdout, stderr = client.exec_command(command)
+        stdout_output = stdout.read().decode("utf-8").strip()
+        stderr_output = stderr.read().decode("utf-8").strip()
+        return_code = stdout.channel.recv_exit_status()
+
+        if log:
+            utils.log(f"Command executed: {command}")
+            utils.log(f"STDOUT: {stdout_output}")
+            utils.log(f"STDERR: {stderr_output}")
+            utils.log(f"Return code: {return_code}")
+
+        return SSHCommandResult(stdout_output, stderr_output, return_code)
+    except Exception as e:
+        if log:
+            utils.log_and_raise_error(
+                f"Failed to execute command over SSH: {str(e)}", Exception
+            )
+        raise
+    finally:
+        client.close()
+
+
+# 449
+def check_write_permissions(cfg: Configs, path: str, log: bool = True) -> bool:
+    """
+    Check if the user has write permissions on the specified path on the remote server.
+    """
+    try:
+        command = f"touch {path}/test_write_permission.txt && rm {path}/test_write_permission.txt"
+        result = run_ssh_command(cfg, command, log=log)
+        return result.returncode == 0
+    except Exception as e:
+        if log:
+            utils.log(f"Write permission check failed: {e}")
+        return False