Read ~/.ssh/config in the SSH pre-check

The paramiko pre-check ignored HostName, Port and IdentityFile, so hosts that rsync's ssh reached through ~/.ssh/config failed before the backup started.

simple_backup/simple_backup.py

@@ -34,7 +34,7 @@ from shutil import get_terminal_size, rmtree, which
 from subprocess import PIPE, STDOUT, Popen
 from tempfile import mkstemp
 from timeit import default_timer
-from typing import ParamSpec, TypedDict, TypeVar
+from typing import Any, ParamSpec, TypedDict, TypeVar
 
 try:
     import paramiko
@@ -469,6 +469,36 @@ class Backup:
             if connected is None:
                 ssh.close()
 
+    def _ssh_config(self, homedir: str) -> dict[str, Any]:
+        """connect() arguments for ssh_host from ~/.ssh/config (HostName, Port, IdentityFile), as rsync's ssh uses it"""
+
+        assert self.ssh_host is not None
+
+        params: dict[str, Any] = {'hostname': self.ssh_host}
+
+        try:
+            host = paramiko.SSHConfig.from_path(f'{homedir}/.ssh/config').lookup(self.ssh_host)
+
+            params['hostname'] = host['hostname']
+
+            if 'port' in host:
+                params['port'] = int(host['port'])
+        except FileNotFoundError:
+            return params
+        except (OSError, paramiko.SSHException, ImportError, ValueError) as e:
+            # ImportError: 'Match exec' needs the invoke module
+            logger.warning('Cannot read %s/.ssh/config: %s', homedir, e)
+
+            return {'hostname': self.ssh_host}
+
+        # Like ssh, skip identity files that don't exist
+        keys = [k for k in host.get('identityfile', []) if os.path.isfile(k)]
+
+        if keys:
+            params['key_filename'] = keys
+
+        return params
+
     def _ssh_login(self, ssh: paramiko.client.SSHClient, homedir: str) -> paramiko.client.SSHClient | None:
         try:
             ssh.load_host_keys(filename=f'{homedir}/.ssh/known_hosts')
@@ -477,11 +507,13 @@ class Backup:
 
         ssh.set_missing_host_key_policy(paramiko.WarningPolicy())
 
+        host = self._ssh_config(homedir)
+
         try:
             # WarningPolicy emits a UserWarning for unknown host keys
             with warnings.catch_warnings():
                 warnings.simplefilter('error', UserWarning)
-                ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT)
+                ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT)
 
             return ssh
         except UserWarning:
@@ -507,7 +539,7 @@ class Backup:
             pass
 
         try:
-            ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT)
+            ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT)
 
             return ssh
         except paramiko.SSHException:
@@ -521,8 +553,7 @@ class Backup:
                     password = getpass(
                         f'{self.ssh_user}@{self.ssh_host}\'s password: ')
 
-                ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT,
+                ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT, password=password)
-                            password=password)
 
                 self._password_auth = True
                 os.environ['SSHPASS'] = password
@@ -545,7 +576,7 @@ class Backup:
             return None
 
         try:
-            ssh.connect(self.ssh_host, username=self.ssh_user, timeout=SSH_TIMEOUT, pkey=pkey)
+            ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT, pkey=pkey)
         except paramiko.SSHException:
             logger.critical('SSH connection to server failed')
 

tests/test_remote.py

@@ -365,7 +365,7 @@ class FakeClient:
         self.policies.append(type(policy).__name__)
 
     def connect(self, hostname: str, **kwargs: Any) -> None:
-        self.connects.append(kwargs)
+        self.connects.append({'hostname': hostname, **kwargs})
         outcome = self.outcomes.pop(0)
 
         if isinstance(outcome, UserWarning):
@@ -404,8 +404,35 @@ class TestLogin:
         assert remote_backup()._ssh_connect(str(tmp_path)) is client
         assert client.known_hosts == f'{tmp_path}/.ssh/known_hosts'
         assert client.connects[0]['timeout'] == sb.SSH_TIMEOUT
+        assert client.connects[0]['hostname'] == 'server'
+        assert 'port' not in client.connects[0]
         assert not client.closed
 
+    def test_ssh_config(self, client: FakeClient, tmp_path: Path) -> None:
+        (tmp_path / '.ssh').mkdir()
+        key = tmp_path / '.ssh' / 'id_backup'
+        key.touch()
+        (tmp_path / '.ssh' / 'config').write_text(
+            f'Host server\n  HostName backup.example.org\n  Port 2222\n'
+            f'  IdentityFile {key}\n  IdentityFile {tmp_path}/.ssh/missing\n')
+        client.outcomes = [ssh_error(), None]
+
+        assert remote_backup()._ssh_connect(str(tmp_path)) is client
+
+        for params in client.connects:
+            assert (params['hostname'], params['port']) == ('backup.example.org', 2222)
+            assert params['key_filename'] == [str(key)]
+
+    def test_bad_ssh_config(self, client: FakeClient, tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
+        (tmp_path / '.ssh').mkdir()
+        (tmp_path / '.ssh' / 'config').write_text('Host server\n  Port abc\n')
+        client.outcomes = [None]
+
+        assert remote_backup()._ssh_connect(str(tmp_path)) is client
+        assert client.connects[0]['hostname'] == 'server'
+        assert 'port' not in client.connects[0]
+        assert 'Cannot read' in caplog.text
+
     @pytest.mark.parametrize('answer, connected', [('y', True), (' Yes', True), ('n', False), ('', False)])
     def test_unknown_host_key(self, client: FakeClient, monkeypatch: pytest.MonkeyPatch,
                               answer: str, connected: bool) -> None: