fix ssh control master check (#3488)

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>
This commit is contained in:
Jörg Thalheim
2025-05-04 12:49:53 +00:00
committed by Mic92
parent a9d1b776e3
commit dc4268dbf2

View File

@@ -1,9 +1,11 @@
# Adapted from https://github.com/numtide/deploykit # Adapted from https://github.com/numtide/deploykit
import errno
import logging import logging
import os import os
import shlex import shlex
import socket import socket
import stat
import subprocess import subprocess
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
@@ -38,27 +40,35 @@ class Host:
ssh_options: dict[str, str] = field(default_factory=dict) ssh_options: dict[str, str] = field(default_factory=dict)
tor_socks: bool = False 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: def __post_init__(self) -> None:
if not self.command_prefix: if not self.command_prefix:
self.command_prefix = self.host self.command_prefix = self.host
if not self.user: if not self.user:
self.user = "root" self.user = "root"
home = Path.home() self.setup_control_master()
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"
def __str__(self) -> str: def __str__(self) -> str:
return self.target return self.target