diff --git a/simple_backup/simple_backup.py b/simple_backup/simple_backup.py index 42c6c0a..fb0d5b2 100755 --- a/simple_backup/simple_backup.py +++ b/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, - password=password) + ssh.connect(**host, username=self.ssh_user, timeout=SSH_TIMEOUT, 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') diff --git a/tests/test_remote.py b/tests/test_remote.py index 73e0f27..9f52097 100644 --- a/tests/test_remote.py +++ b/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: