From dc4268dbf251a48c39d70c2bbf5c6155d6f103db Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?J=C3=B6rg=20Thalheim?= <joerg@thalheim.io>
Date: Sun, 4 May 2025 12:49:53 +0000
Subject: [PATCH] fix ssh control master check (#3488)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: pinpox <git@pablo.tools>
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/3488
Co-authored-by: Jörg Thalheim <joerg@thalheim.io>
Co-committed-by: Jörg Thalheim <joerg@thalheim.io>
---
 pkgs/clan-cli/clan_cli/ssh/host.py | 42 ++++++++++++++++++------------
 1 file changed, 26 insertions(+), 16 deletions(-)

diff --git a/pkgs/clan-cli/clan_cli/ssh/host.py b/pkgs/clan-cli/clan_cli/ssh/host.py
index 12d2ff4d5..97f6a619c 100644
--- a/pkgs/clan-cli/clan_cli/ssh/host.py
+++ b/pkgs/clan-cli/clan_cli/ssh/host.py
@@ -1,9 +1,11 @@
 # Adapted from https://github.com/numtide/deploykit
 
+import errno
 import logging
 import os
 import shlex
 import socket
+import stat
 import subprocess
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -38,27 +40,35 @@ class Host:
     ssh_options: dict[str, str] = field(default_factory=dict)
     tor_socks: bool = False
 
+    def setup_control_master(self) -> None:
+        home = Path.home()
+        if not home.exists():
+            return
+        control_path = home / ".ssh"
+        try:
+            if not stat.S_ISDIR(control_path.stat().st_mode):
+                return
+        except OSError as e:
+            if e.errno == errno.ENOENT:
+                try:
+                    control_path.mkdir(exist_ok=True)
+                except OSError:
+                    return
+            else:
+                return
+
+        self.ssh_options["ControlMaster"] = "auto"
+        # Can we make this a temporary directory?
+        self.ssh_options["ControlPath"] = str(control_path / "clan-%h-%p-%r")
+        # We use a short ttl because we want to mainly re-use the connection during the cli run
+        self.ssh_options["ControlPersist"] = "1m"
+
     def __post_init__(self) -> None:
         if not self.command_prefix:
             self.command_prefix = self.host
         if not self.user:
             self.user = "root"
-        home = Path.home()
-        if home.exists() and os.access(home, os.W_OK):
-            control_path = home / ".ssh"
-            if not control_path.exists():
-                try:
-                    control_path.mkdir(exist_ok=True)
-                except OSError:
-                    pass
-                else:
-                    self.ssh_options["ControlMaster"] = "auto"
-                    # Can we make this a temporary directory?
-                    self.ssh_options["ControlPath"] = str(
-                        control_path / "clan-%h-%p-%r"
-                    )
-                    # We use a short ttl because we want to mainly re-use the connection during the cli run
-                    self.ssh_options["ControlPersist"] = "1m"
+        self.setup_control_master()
 
     def __str__(self) -> str:
         return self.target