carletes / mock-ssh-server

Python mock SSH server for testing purposes

mock-ssh-server - An SSH server for testing purposes

mock-ssh-server packs a Python context manager that implements an SSH server for testing purposes. It is built on top of paramiko, so it does not need OpenSSH binaries to be installed.

Sample usage

As a py.test fixture:

import os

from pytest import yield_fixture

import mockssh


@yield_fixture()
def server():
    users = {
        "sample-user": "/path/to/user-private-key",
    }
    with mockssh.Server(users) as s:
        yield s


def test_ssh_session(server):
    for uid in server.users:
        with server.client(uid) as c:
            _, stdout, _ = c.exec_command("ls /")
            assert stdout.read()

def test_sftp_session(server):
    for uid in server.users:
        target_dir = tempfile.mkdtemp()
        target_fname = os.path.join(target_dir, "foo")
        assert not os.access(target_fname, os.F_OK)

        with server.client(uid) as c:
            sftp = c.open_sftp()
            sftp.put(__file__, target_fname, confirm=True)
            assert os.access(target_fname, os.F_OK)

mock-ssh-server's Issues

setup.py version is 0.4.0 while pypi is 0.5.0

Hello! I'm wondering if you can release a 0.5.1 version including these changes: #9

Also, I noticed that the version specified on the setup.py don't match the one in https://pypi.org/project/mock-ssh-server/#history

Windows support is broken for executing commands received in bytes

I am not sure if there is an explicit windows support (since the CI only tests Linux), but when executing commands it fails with this error;

INFO     asyncssh:logging.py:79 [conn=0, chan=1]   Command: cp '\Users\RUNNER~1\AppData\Local\Temp\dvc-test.6736.o8cti5un.KpcCKHkfHUgi4gWYYHpDcK\.nb8p42n9d235B6AdAhbcQR.tmp' '\Users\RUNNER~1\AppData\Local\Temp\dvc-test.6736.o8cti5un.KpcCKHkfHUgi4gWYYHpDcK\foo'
DEBUG    mockssh.server:server.py:57 Executing b"cp '\\Users\\RUNNER~1\\AppData\\Local\\Temp\\dvc-test.6736.o8cti5un.KpcCKHkfHUgi4gWYYHpDcK\\.nb8p42n9d235B6AdAhbcQR.tmp' '\\Users\\RUNNER~1\\AppData\\Local\\Temp\\dvc-test.6736.o8cti5un.KpcCKHkfHUgi4gWYYHpDcK\\foo'"
ERROR    mockssh.server:server.py:67 Error handling client (channel: <paramiko.Channel 1 (open) window=2097152 -> <paramiko.Transport at 0x430e80a0 (cipher aes256-ctr, 256 bits) (active; 2 open channel(s))>>)
Traceback (most recent call last):
  File "C:\hostedtoolcache\windows\Python\3.9.6\x64\lib\site-packages\mockssh\server.py", line 58, in handle_client
    p = subprocess.Popen(command, shell=True,
  File "C:\hostedtoolcache\windows\Python\3.9.6\x64\lib\subprocess.py", line 951, in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
  File "C:\hostedtoolcache\windows\Python\3.9.6\x64\lib\subprocess.py", line 1352, in _execute_child

I was able to get around it with subclassing the Handler, and adding a .decode() to the handle_client() method. If it is applicable, I can convert it to a PR and submit here.

Sech channel 0 open FAILED: Administratively prohibited

Hi,
I'm using mock-ssh-server v0.9.1 + sample-user:sample-user-key for port forwarding.

I receive the following error: Sech channel 0 open FAILED: Administratively prohibited.

How to configure mock to allow TCP PORT Forwarding ?

Thanks

Michael.

`stat` on SFTPHandle causes paramiko prefetch mechanism to fail

I have a client use case where I do:

with self._client.open(file_name, mode) as file:
    file.prefetch()
    # rest of file processing here

Unfortunately paramiko internally calls: resp = self.file_table[handle].stat() where self.file_table[handle] return instance of mockssh.sftp.SFTPHandle that lacks this method.

I now patch my tests with following stat() method implementation:

def stat(self):
    st = os.fstat(self.file_obj.name)
    return paramiko.SFTPAttributes.from_stat(st)

Maybe we could make it the default or support it in any other way?

Getting OSError: [Errno 9] Bad file descriptor in tests

I occasionally get:

Exception in thread Thread-9:
Traceback (most recent call last):
  File "/usr/lib/python3.7/threading.py", line 926, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.7/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "/home/suor/.virtualenvs/dvc/lib/python3.7/site-packages/mockssh/server.py", line 126, in _run
    conn, addr = sock.accept()
  File "/usr/lib/python3.7/socket.py", line 212, in accept
    fd, addr = self._accept()
OSError: [Errno 9] Bad file descriptor

which doesn't break anything, only annoys us.

Here is how we use it in our pytest suite:

here = os.path.abspath(os.path.dirname(__file__))

user = "user"
key_path = os.path.join(here, "{0}.key".format(user))


@pytest.fixture
def ssh_server():
    users = {user: key_path}
    with mockssh.Server(users) as s:
        s.test_creds = {
            "host": s.host,
            "port": s.port,
            "username": user,
            "key_filename": key_path,
        }
        yield s


@pytest.fixture
def ssh(ssh_server):
    # This is our class encapsulating paramiko.SSHClient
    yield SSHConnection(**ssh_server.test_creds)

stdin of executed command is not closed when using the `ssh` binary as the client

Since I'm currently trying to write a test harness for some legacy ssh scripts with the basic idea cat long-input-file | ssh remotehost process.sh, I'm really happy that mockssh just recently gained stdin support in 0.9.1. Thank you!

However, it seems to me that paramiko's server.client().execute_command() handles stdin differently than openssh's ssh binary. This example test hangs indefinitely at while self.process.poll():

@pytest.fixture
def server(tmpdir):
    shutil.copy(pkg_resources.resource_filename(
        'mockssh', 'sample-user-key'), tmpdir / 'key')
    os.chmod(tmpdir / 'key', 0o600)  # ssh(1) requires strict permissions
    with mockssh.Server({'sample': pkg_resources.resource_filename(
            'mockssh', 'sample-user-key')}) as server:
        yield server

def test_stdin_using_ssh_client_binary(server, tmpdir):
    proc = subprocess.Popen(
        ['ssh', '-i', tmpdir / 'key', '-p', str(server.port), 
         '-o', 'StrictHostKeyChecking=no',
         'sample@localhost',  'cat'],
        stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err = proc.communicate(b'myinput')
    out, err = out.decode('utf-8'), err.decode('utf-8')
    print(err)
    assert 'myinput' in out

I'm not at all familiar with all the internals at play here, so all I can offer is this somewhat kludgy idea -- which does make the above test case work as expected:

--- mockssh/streaming.py~	2021-10-25 14:03:25.000000000 +0200
+++ mockssh/streaming.py	2021-10-25 14:04:20.000000000 +0200
@@ -23,6 +23,23 @@
                 return


+class InputStream(Stream):
+
+    def __init__(self, fd, read, output):
+        super().__init__(fd, read, output.write, output.flush)
+        self.output = output
+        self.eof = False
+
+    def transfer(self):
+        if self.eof:
+            return b''
+        data = super().transfer()
+        # https://docs.paramiko.org/en/stable/api/channel.html#paramiko.channel.Channel.recv says
+        # "If a string of length zero is returned, the channel stream has closed."
+        if not data:
+            self.eof = True
+            self.output.close()
+        return data
+
+
 class StreamTransfer:
     BUFFER_SIZE = 1024

@@ -35,7 +52,7 @@
         ]

     def ssh_to_process(self, channel, process_stream):
-        return Stream(channel, lambda: channel.recv(self.BUFFER_SIZE), process_stream.write, process_stream.flush)
+        return InputStream(channel, lambda: channel.recv(self.BUFFER_SIZE), process_stream)

     @staticmethod
     def process_to_ssh(process_stream, write_func):

Docs - remove reference to @yield_fixture

The README references @yield_fixture(), this has been deprecated since pytest 3, which came out in 2016.

Quick Question - Iris

Hey I have founded a company called Iris, which is an app that notifies your friends every time you make a trade on Robinhood, or whatever broker you use. We are VC-backed and have 9 engineers. We are looking to hire a python engineer. We were building out a python monorepo and found one of your repos.

Do you have time for a 30 min call?

rsync hangs

For test purposes, I tried to use rsync to transfer files but it always hang. I printed the log and found it was stuck when rsync tried to call ssh to start a remote client.

Is it a limitation on the mock ssh server?

support for authenticating with passwords instead of ssh keys

It would be really if the mockssh.Server could support authenticating with other methods as well (primarily passwords).