fix wrong error raised when private key is missing or invalid

sftpretty/__init__.py

@@ -187,8 +187,11 @@ class Connection(object):
     :raises ConnectionException:
     :raises CredentialException:
     :raises HostKeysException:
+    :raises KeyError:
     :raises LoggingException:
+    :raises OSError:
     :raises PasswordRequiredException:
+    :raises PermissionError:
     :raises SSHException:
     '''
     def __init__(self, host, cnopts=None, default_path=None, password=None,
@@ -214,33 +217,39 @@ def _set_authentication(self, password, private_key, private_key_pass):
         if private_key is not None:
             # Use key path or provided key object
             key_types = {'EC': ECDSAKey, 'OPENSSH': Ed25519Key, 'RSA': RSAKey}
-            if isinstance(private_key, str):
+            if isinstance(private_key, (str, Path)):
                 key_file = Path(private_key).expanduser().absolute().as_posix()
                 try:
-                    with open(key_file, 'r', encoding='utf-8') as head:
-                        key_id = head.readline()[11:][:-18]
+                    with open(key_file, 'rb') as head:
+                        header = head.readline(64).decode('ascii', 'replace')
+                    key_id = header.rpartition(' PRIVATE KEY-----')[0][11:]
                     log.debug(f'Key ID: [{key_id}]')
                     key = key_types[key_id.strip()]
-                except KeyError as err:
-                    log.error(('Unable to identify key type from file provided'
-                              f': \n[{key_file}]'))
-                    raise err
-                except PasswordRequiredException as err:
-                    log.error(('No password provided for encrypted private '
-                               'key encrypted private key.'))
-                    raise err
-                except PermissionError as err:
-                    log.error(('File permission preventing user access to:\n'
-                              f'[{key_file}]'))
-                    raise err
-                except SSHException as err:
-                    log.error(('Path provided is an invalid key file, a '
-                               'directory or does not exist, please revise '
-                               'and provide a path to a valid private key.'))
-                    raise err
-                finally:
                     private_key = key.from_private_key_file(
                         key_file, password=private_key_pass)
+                except KeyError:
+                    log.error(('Unsupported key format, paramiko only reads '
+                               'EC, OPENSSH and RSA PEM keys. Re-encode with '
+                              f'ssh-keygen -p -f <keyfile>:\n[{key_file}]'))
+                    raise
+                except PermissionError:
+                    log.error(('File permission preventing user access to:\n'
+                              f'[{key_file}]'))
+                    raise
+                except OSError:
+                    log.error(('Path provided is a directory or does not '
+                               'exist, please revise and provide a path to a '
+                              f'readable private key:\n[{key_file}]'))
+                    raise
+                except PasswordRequiredException:
+                    log.error(('No password provided for encrypted private '
+                               f'key:\n[{key_file}]'))
+                    raise
+                except SSHException:
+                    log.error(('Path provided is an invalid or corrupt key '
+                               'file, please revise and provide a path to a '
+                               'valid private key.'))
+                    raise
             self._transport.auth_publickey(self._username, private_key)
         elif password is not None:
             self._transport.auth_password(self._username, password)

tests/common.py

@@ -6,13 +6,16 @@
 from os import close, environ
 from pathlib import Path
 from sftpretty import CnOpts
+from stat import S_ISDIR
 from tempfile import mkstemp
 
 
 PASS = 'tEst@!357'
 SKIP_IF_CI = pytest.mark.skipif(environ.get('CI', '') > '', reason='Not Local')
 SKIP_IF_MAC = pytest.mark.skipif(environ.get('RUNNER_OS', '') == 'macOS',
                                  reason='WhackMac')
+SKIP_IF_ROOT = pytest.mark.skipif(environ.get('USER', '') == 'root',
+                                  reason='RootRules')
 SKIP_IF_WIN = pytest.mark.skipif(environ.get('RUNNER_OS', '') == 'Windows',
                                  reason='NoWinZone')
 STARS8192 = '*' * 8192
@@ -36,7 +39,25 @@ def conn(sftpsrv):
             'username': USER}
 
 
+def remote_rmdir(sftp, dir):
+    '''recursively remove a remote directory tree'''
+    try:
+        listing = sftp.listdir_attr(dir)
+    except FileNotFoundError:
+        return
+
+    for attr in listing:
+        remotepath = Path(dir).joinpath(attr.filename).as_posix()
+        if S_ISDIR(attr.st_mode):
+            remote_rmdir(sftp, remotepath)
+        else:
+            sftp.remove(remotepath)
+
+    sftp.rmdir(dir)
+
+
 def rmdir(dir):
+    '''recursively remove a directory tree'''
     dir = Path(dir)
     for item in dir.iterdir():
         if item.is_dir():

tests/conftest.py

@@ -4,8 +4,9 @@
 
 from paramiko.hostkeys import HostKeys
 from pathlib import Path
+from uuid import uuid4
 
-from common import LOCAL
+from common import LOCAL, remote_rmdir, USER_HOME
 from sftpretty import CnOpts, Connection
 
 
@@ -16,6 +17,7 @@ def lsftp(request):
     LOCAL['cnopts'] = cnopts
     lsftp = Connection(**LOCAL)
     request.addfinalizer(lsftp.close)
+
     return lsftp
 
 
@@ -36,3 +38,15 @@ def knownhosts(sftpserver, key_type='ssh-ed25519'):
     knownhosts.write_bytes(bytes(hostkeys, 'utf-8'))
 
     return
+
+
+@pytest.fixture
+def remote_tmpdir(lsftp):
+    '''setup unique remote temporary directory'''
+    remotedir = Path(USER_HOME).joinpath(f'sftpretty-{uuid4().hex[:8]}')
+    lsftp.mkdir_p(remotedir.as_posix())
+
+    try:
+        yield remotedir.as_posix()
+    finally:
+        remote_rmdir(lsftp, remotedir.as_posix())

tests/test_connection.py

@@ -94,6 +94,41 @@ def test_connection_bad_host():
         sftp.listdir()
 
 
+@pytest.mark.parametrize('blob', (
+    b'\x30\x82\x04\xbe\x02\x01\x00',             # binary DER, undecodable
+    b'-----BEGIN DSA PRIVATE KEY-----\n',        # deprecated algorithm
+    b'-----BEGIN ENCRYPTED PRIVATE KEY-----\n',  # PKCS#8, encrypted
+    b'-----BEGIN PRIVATE KEY-----\n',            # PKCS#8
+    b''                                          # empty file
+))
+def test_connection_bad_private_key_format(blob, tmp_path):
+    '''deprecated or unsupported key formats must raise, not fail'''
+    key = tmp_path.joinpath('id_sftpretty_unsupported')
+    key.write_bytes(blob)
+
+    copts = LOCAL.copy()
+    copts['private_key'] = key.as_posix()
+    with pytest.raises(KeyError):
+        with Connection(**copts) as sftp:
+            sftp.listdir()
+
+
+@pytest.mark.parametrize('kind', ('missing', 'directory'))
+def test_connection_bad_private_key_path(kind, tmp_path):
+    '''private-key path pointing to missing or non-file type'''
+    key = tmp_path.joinpath(f'id_sftpretty_{kind}')
+
+    if kind == 'directory':
+        key.mkdir()
+
+    copts = LOCAL.copy()
+    copts['private_key'] = key.as_posix()
+
+    with pytest.raises(OSError, match=key.name):
+        with Connection(**copts) as sftp:
+            sftp.listdir()
+
+
 def test_connection_good(sftpserver):
     '''connect to a public sftp server'''
     with sftpserver.serve_content(VFS):

tests/test_normalize.py

@@ -1,8 +1,10 @@
 '''test sftpretty.normalize'''
 
 from common import VFS, conn
+from io import BytesIO
 from pathlib import Path
 from sftpretty import Connection
+from stat import S_ISLNK
 
 
 def test_normalize(sftpserver):
@@ -18,13 +20,27 @@ def test_normalize(sftpserver):
             assert sftp.normalize('.') == pubpath.as_posix()
 
 
-# TODO
-# def test_normalize_symlink(sftp):
-#     '''test normalize against a symlink'''
-#     home = Path.home()
-#     sftp.chdir(home.as_posix())
-#     rsym = 'readme.sym'
-#     assert sftp.normalize(rsym) == home.joinpath(rsym).as_posix()
+def test_normalize_dangling_symlink(lsftp, remote_tmpdir):
+    '''test normalize against a symlink whose target is missing'''
+    missing = Path(remote_tmpdir).joinpath('gone.txt').as_posix()
+    rsym = Path(remote_tmpdir).joinpath('dangling.sym').as_posix()
+    lsftp.symlink(missing, rsym)
+
+    assert lsftp.lexists(rsym)
+    assert lsftp.exists(rsym) is False
+    assert lsftp.normalize(rsym) == missing
+
+
+def test_normalize_symlink(lsftp, remote_tmpdir):
+    '''test normalize against a symlink'''
+    rfile = Path(remote_tmpdir).joinpath('readme.txt').as_posix()
+    rsym = Path(remote_tmpdir).joinpath('readme.sym').as_posix()
+    lsftp.putfo(BytesIO(b'My hovercraft is full of eels.'), rfile)
+    lsftp.symlink(rfile, rsym)
+
+    assert S_ISLNK(lsftp.lstat(rsym).st_mode)
+    assert lsftp.normalize(rsym) == lsftp.normalize(rfile)
+    assert lsftp.normalize(rsym) != rsym
 
 
 def test_pwd(sftpserver):

tests/test_put.py

@@ -46,6 +46,7 @@ def test_put_callback(lsftp):
         lsftp.put(fname, callback=cback)
         # clean up
         lsftp.remove(base_fname)
+
     # verify callback was called
     assert cback.call_count
 
@@ -58,6 +59,7 @@ def test_put_confirm(lsftp):
         result = lsftp.put(fname)
         # clean up
         lsftp.remove(base_fname)
+
     # verify that an SFTPAttribute like Path.stat() was returned
     assert result.st_size == 8192
     assert result.st_uid is not None
@@ -67,11 +69,11 @@ def test_put_confirm(lsftp):
 
 
 # TODO
-# def test_put_not_allowed(psftp):
+# def test_put_not_allowed(lsftp):
 #     '''try to put a file to a read-only server'''
 #     with tempfile_containing() as fname:
 #         with pytest.raises(IOError):
-#             psftp.put(fname)
+#             lsftp.put(fname)
 
 
 def test_put_preserve_mtime(lsftp):
@@ -85,6 +87,7 @@ def test_put_preserve_mtime(lsftp):
         result2 = lsftp.put(fname, preserve_mtime=True)
         # clean up
         lsftp.remove(base_fname)
+
     # see if times are modified
     # assert base.st_atime == result1.st_atime
     assert int(base.st_mtime) == result1.st_mtime
@@ -101,5 +104,6 @@ def test_put_resume(lsftp):
         with open(fname, 'ab') as fh:
             fh.write('this...'.encode('utf-8'))
         result = lsftp.put(fname, preserve_mtime=True, resume=True)
+
     assert base.st_size == result.st_size
     assert partial.st_mtime == result.st_mtime

tests/test_put_d.py

@@ -3,32 +3,42 @@
 import pytest
 
 from blddirs import build_dir_struct
-from common import rmdir
+from common import SKIP_IF_ROOT
 from pathlib import Path
-from tempfile import mkdtemp
 
 
-def test_put_d(lsftp):
+def test_put_d(lsftp, remote_tmpdir, tmp_path):
     '''test put_d'''
-    localpath = Path(mkdtemp()).as_posix()
-    remote = Path.home()
-    build_dir_struct(localpath)
-    local = Path(localpath).joinpath('pub')
-    lsftp.put_d(local.as_posix(), remote.as_posix())
+    build_dir_struct(tmp_path.as_posix())
+    local = tmp_path.joinpath('pub').as_posix()
+    lsftp.put_d(local, remote_tmpdir)
+    remote = Path(remote_tmpdir).joinpath('pub').as_posix()
 
-    rmdir(localpath)
+    assert lsftp.listdir(remote) == ['make.txt']
 
 
-# TODO
-# def test_put_d_ro(psftp):
-#     '''test put_d failure on remote read-only srvr'''
-#     # run the op
-#     with pytest.raises(IOError):
-#         psftp.put_d('.', '.')
+@SKIP_IF_ROOT
+@pytest.mark.parametrize('refuse', ('mkdir', 'write'))
+def test_put_d_ro(lsftp, refuse, remote_tmpdir, tmp_path):
+    '''test put_d failure on remote read-only server'''
+    build_dir_struct(tmp_path.as_posix())
+    local = tmp_path.joinpath('pub').as_posix()
 
+    if refuse == 'mkdir':
+        remote = remote_tmpdir
+    else:
+        remote = Path(remote_tmpdir).joinpath('pub').as_posix()
+        lsftp.mkdir_p(remote)
 
-def test_put_d_bad_local(lsftp):
+    lsftp.chmod(remote, 500)
+    try:
+        with pytest.raises(PermissionError):
+            lsftp.put_d(local, remote_tmpdir)
+    finally:
+        lsftp.chmod(remote, 700)
+
+
+def test_put_d_bad_local(lsftp, remote_tmpdir):
     '''test put_d failure on non-existing local directory'''
-    # run the op
     with pytest.raises(OSError):
-        lsftp.put_d('/non-existing', '.')
+        lsftp.put_d('/non-existing', remote_tmpdir)

tests/test_put_r.py

@@ -3,32 +3,25 @@
 import pytest
 
 from blddirs import build_dir_struct
-from common import rmdir
-from pathlib import Path
-from tempfile import mkdtemp
 
 
-def test_put_r(lsftp):
+def test_put_r(lsftp, remote_tmpdir, tmp_path):
     '''test put_r'''
-    localpath = Path(mkdtemp()).as_posix()
-    remote = Path.home()
-    build_dir_struct(localpath)
-    local = Path(localpath).joinpath('pub')
-    lsftp.put_r(local.as_posix(), remote.as_posix())
+    build_dir_struct(tmp_path.as_posix())
+    local = tmp_path.joinpath('pub').as_posix()
+    lsftp.put_r(local, remote_tmpdir)
 
-    rmdir(localpath)
+    assert lsftp.listdir(remote_tmpdir) != []
 
 
 # TODO
-# def test_put_r_ro(psftp):
-#     '''test put_r failure on remote read-only srvr'''
-#     # run the op
+# def test_put_r_ro(lsftp):
+#     '''test put_r failure on remote read-only server'''
 #     with pytest.raises(IOError):
-#         psftp.put_r('.', '.')
+#         lsftp.put_r('.', '.')
 
 
-def test_put_r_bad_local(lsftp):
+def test_put_r_bad_local(lsftp, remote_tmpdir):
     '''test put_r failure on non-existing local directory'''
-    # run the op
     with pytest.raises(OSError):
-        lsftp.put_r('/non-existing', '.')
+        lsftp.put_r('/non-existing', remote_tmpdir)