Skip to content
Snippets Groups Projects
Commit 52dd4c16 authored by Georges Racinet's avatar Georges Racinet
Browse files

SSH test helpers

This provides facilities to start SSH servers constrained to
a single user and use them from `hg` commands, complete with host
and user key pairs.

`hgitaly.testing.tests.test_sshd` demonstrates a successful push
using them, allowing the forthcoming tests for service methods using
SSH to focus on their own business.
parent 5e726a46
No related branches found
No related tags found
1 merge request!89MercurialRepositoryService.Push
# Copyright 2021 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
#
# SPDX-License-Identifier: GPL-2.0-or-later
from contextlib import contextmanager
from tempfile import NamedTemporaryFile
@contextmanager
def client_command(priv_key, known_hosts):
"""Context manager providing a suitable ssh client command.
The command references temporary files that get cleaned up
automatically when exiting the context.
:param str priv_key: non-encrypted private key file content.
"""
with (
NamedTemporaryFile(mode='w', delete=False) as keyf,
NamedTemporaryFile(mode='w', delete=False) as known_hostsf,
):
keyf.write(priv_key)
keyf.close()
known_hostsf.write(known_hosts)
known_hostsf.close()
yield ['ssh',
'-o', 'IdentitiesOnly=yes',
'-o', 'UserKnownHostsFile=%s' % known_hostsf.name,
# do not prompt user if server key is unknown
'-o', 'StrictHostKeyChecking=yes',
'-i', keyf.name
]
@contextmanager
def client_command_str(*args, **kwargs):
with client_command(*args, **kwargs) as cmd:
yield ' '.join(cmd)
# Copyright 2021 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
#
# SPDX-License-Identifier: GPL-2.0-or-later
import pathlib
TEST_DATA_DIR = pathlib.Path(__file__).parent / 'data'
ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQC0OxPfeeWo979RwTz5iMLfZttkWQHBCP0LZKNkEtu4gfDLzcg+z7crJSVdPTbY3k6ThqMicQUxPFHM77bb+EvQjkIE03WpehSutXfIYiSVkECw7MZpCB+/VbZTRClI7lQEnrKeoxiQABylArtWtyH7yCIMeHT2gkbpmbgoEtTJxiPl3t5ccvSblqqLQGhuNQ5iaOzvpB8lFZjBHa8jbOAHTDm8jgRL/m9AKKcynHiKb4WAb4KozI/I0Bu5uS6dW7oGxmM1g0boGOd9l52ZYXDnRvBvnEzyEAjEMtArm9c1L79ugNmcdOs1jUWqfYHEBkBHKqDpbU88uqeN/WPTvPTYLmn5pNhSbRFuEymbIjRnEVnJsfQhCNfFiqY7bfMvBrXarGm9mwJMLZIiPm07qmApgGxp1a58ME9PIiU2r+9gLesxCtTuHRfCFhomny6psaaOBl6rasOhYITMuPVov1u5WULDKctMAL9s4570hWUawNZzHOsLnEZiDeqbGYU1Yl8= hgitaly-tests@localhost
-----BEGIN OPENSSH PRIVATE KEY-----
b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABlwAAAAdzc2gtcn
NhAAAAAwEAAQAAAYEAtDsT33nlqPe/UcE8+YjC32bbZFkBwQj9C2SjZBLbuIHwy83IPs+3
KyUlXT022N5Ok4ajInEFMTxRzO+22/hL0I5CBNN1qXoUrrV3yGIklZBAsOzGaQgfv1W2U0
QpSO5UBJ6ynqMYkAAcpQK7Vrch+8giDHh09oJG6Zm4KBLUycYj5d7eXHL0m5aqi0BobjUO
Ymjs76QfJRWYwR2vI2zgB0w5vI4ES/5vQCinMpx4im+FgG+CqMyPyNAbubkunVu6BsZjNY
NG6BjnfZedmWFw50bwb5xM8hAIxDLQK5vXNS+/boDZnHTrNY1Fqn2BxAZARyqg6W1PPLqn
jf1j07z02C5p+aTYUm0RbhMpmyI0ZxFZybH0IQjXxYqmO23zLwa12qxpvZsCTC2SIj5tO6
pgKYBsadWufDBPTyIlNq/vYC3rMQrU7h0XwhYaJp8uqbGmjgZeq2rDoWCEzLj1aL9buVlC
wynLTAC/bOOe9IVlGsDWcxzrC5xGYg3qmxmFNWJfAAAFiEcrvIZHK7yGAAAAB3NzaC1yc2
EAAAGBALQ7E9955aj3v1HBPPmIwt9m22RZAcEI/Qtko2QS27iB8MvNyD7PtyslJV09Ntje
TpOGoyJxBTE8Uczvttv4S9COQgTTdal6FK61d8hiJJWQQLDsxmkIH79VtlNEKUjuVASesp
6jGJAAHKUCu1a3IfvIIgx4dPaCRumZuCgS1MnGI+Xe3lxy9JuWqotAaG41DmJo7O+kHyUV
mMEdryNs4AdMObyOBEv+b0AopzKceIpvhYBvgqjMj8jQG7m5Lp1bugbGYzWDRugY532XnZ
lhcOdG8G+cTPIQCMQy0Cub1zUvv26A2Zx06zWNRap9gcQGQEcqoOltTzy6p439Y9O89Ngu
afmk2FJtEW4TKZsiNGcRWcmx9CEI18WKpjtt8y8Gtdqsab2bAkwtkiI+bTuqYCmAbGnVrn
wwT08iJTav72At6zEK1O4dF8IWGiafLqmxpo4GXqtqw6FghMy49Wi/W7lZQsMpy0wAv2zj
nvSFZRrA1nMc6wucRmIN6psZhTViXwAAAAMBAAEAAAGASk6BJwYO1dcvHfnZtdfY4Fp74a
eAvaGDEotvgfb0d+G68YvqH1tsXXF3MjXmwwsEwnYkFDMcvOvtndzHO3HncBj2xx6XyEP4
+v7+X7N2+eL7iPZV5z/eNviTr0jO7K62HSGA6z67BcPGC5JQwrPGYxRkVVfkgS29Jtqlvn
/lEzuJ2WBFtT//8ZXwwnDPnYBqadBwv2yfGp8I17XxW12sD9Yxh7fVta81FvKr3IX+fLT2
Cbnv2MBeF6EVjUD4IxCJtL4pZxVKXPni1+lJ9E06DbNeSlJzx39kkabXlzK8vXYbdNbf44
JM9pQiSl+hr0G+SfjnQK+eSle/+DmvF3UoajqtvdW0qhqESeDPNUipybFPtCS7OUqeRzQL
6yjixTzJNYmpQxI6XTHguE4KpBHjO/Rsrl0ghU+c5c0lX7J9Lg+Wuq7HQAfs0xMF8MU48t
x0JFIy1dSHAv8Xf7cnUU2PXJct6oat9fqt+zwowrd0CcaI6/m5U6ULexElLn+F+pRBAAAA
wBD/L+aElfZS5eSa8NG3gzEtAjEH/yPYLsZ2QAyfyoLDqysvtoatHw0BIuVwZCSeSZBd17
K0DBaHHK9DycRSzN26Lu2kx4nV9Qlo6MBscTDg4c/NCMPNF3N1xIf0XOja2j+4i8Bzqkix
/ItDS5LkaQWREQI6/ijHzoKSAyZb3Np1lQGWVv2Ew6IzIr/8/e8f/tbHRYnzzXpFr61JbB
UlyYdTazjDPdJ9JIXt09EJwooYRm4iP/wSMwGCq8WGod6MCQAAAMEA3oSCGAU5pko5DzId
MHCnmvMWf88QMUhCTN0rQa4F1HgbvCTAIbpc2urcqGBXB0xQolZmFMSrowyNFh+nH2DRgC
MMGzhg7hxYrZ6dDbifXv2OVcE6Ln24dpd3k0Ohvk1A6Zq07sEMo/ii6774FelXGhnr1AUQ
QgA2QVu6rAAEL6x1hb4e2to9oOWIlnjUiaKKtpyr7wijIkLXqe95RB9RRH3SAHZCVA8aja
AZ2eMQiK5sq7FjfHmyPybvqYdCg43nAAAAwQDPWaisauZQgmULQSPPdkc0efBRfs6HEd1s
H1MmFrXS/YOu2eAnv+UrcPH8OLfXaW0qDDF+Jg2BmHrOB7wC9Su5o9R7+GZDiptgkDREub
6yZhh8RhQMozq6q+ebHu40IUYT8iYT5/fcUGRpBwc5RDTaRVXseF1StN77HCYgzUf/APp6
xq+k1akGBLYPiE09OiHcIMC4S9935TkHNc5giImydHx0b8Tg+MTqYqwJKUItsawdRYFx0U
vJzGlX7T+uSMkAAAASZ3JhY2luZXRAbXV0YXRpb25zAQ==
-----END OPENSSH PRIVATE KEY-----
[{host}]:{port} ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBK9JxbS6PkBHi4W08qmDyTx7l8ahR5ftxnokWR+Aa1GvuQLKiezakYKSs8iwF1jLQMEI5LazUOdHiPyfuh2OzY0=
-----BEGIN OPENSSH PRIVATE KEY-----
b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAAAaAAAABNlY2RzYS
1zaGEyLW5pc3RwMjU2AAAACG5pc3RwMjU2AAAAQQSvScW0uj5AR4uFtPKpg8k8e5fGoUeX
7cZ6JFkfgGtRr7kCyons2pGCkrPIsBdYy0DBCOS2s1DnR4j8n7odjs2NAAAAsBYr1JMWK9
STAAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBK9JxbS6PkBHi4W0
8qmDyTx7l8ahR5ftxnokWR+Aa1GvuQLKiezakYKSs8iwF1jLQMEI5LazUOdHiPyfuh2OzY
0AAAAgTJswdWy+Yi/En5Fj+IJzZNUk+DHsgc54PDNqrW5JvbUAAAASZ3JhY2luZXRAbXV0
YXRpb25zAQIDBAUG
-----END OPENSSH PRIVATE KEY-----
ecdsa-sha2-nistp256 AAAAE2VjZHNhLXNoYTItbmlzdHAyNTYAAAAIbmlzdHAyNTYAAABBBK9JxbS6PkBHi4W08qmDyTx7l8ahR5ftxnokWR+Aa1GvuQLKiezakYKSs8iwF1jLQMEI5LazUOdHiPyfuh2OzY0= gracinet@mutations
AllowUsers {user}
AuthenticationMethods publickey
#HostKey /home/gracinet/heptapod/hdk/dev-0.27/openssh/ssh_host_rsa_key
ListenAddress {host}:{port}
LogLevel VERBOSE
PasswordAuthentication no
PidFile none
Protocol 2
StrictModes no
UseLogin no
UsePAM no
UsePrivilegeSeparation no
AcceptEnv GIT_PROTOCOL
Match User {user}
AuthorizedKeysFile {authorized_keys_path}
# Copyright 2021 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
#
# SPDX-License-Identifier: GPL-2.0-or-later
import os
from pathlib import Path
import sys
def hg_exe_path(): # pragma no cover
"""Return a Mercurial executable path consistent with these tests.
The executable should be consistent with the `mercurial` package imported
in the tests.
Most of these tests don't need to invoke Mercurial subprocesses, but
some will have to (SSH push). In that case, we need a path to a version
that is equivalent enough. A simple heuristic to achieve that is to try and
return the path to the Mercurial executable that was installed with the
`mercurial` package (e.g. from the same virtualenv etc.).
If that is not enough, the `HGITALY_TESTS_HG_EXE_PATH` environment
variable can be set to something suitable.
"""
from_env = os.environ.get('HGITALY_TESTS_HG_EXE_PATH')
if from_env:
return from_env
return str(Path(sys.executable).parent / 'hg')
# Copyright 2021 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
#
# SPDX-License-Identifier: GPL-2.0-or-later
import attr
import os
import pwd
import socket
import subprocess
from . import TEST_DATA_DIR
@attr.s
class SSHServer:
host_pub_key_path = attr.ib()
host_priv_key_path = attr.ib()
config_template_path = attr.ib()
working_dir = attr.ib()
authorized_keys_path = attr.ib()
host_ipv4 = attr.ib(default='127.0.0.1')
"""IPv4 local address for the server to bind to.
Domain names and IPv6 addresses are not supported (not even `::1`).
Forcing the communication to be over IPv4 is lame, but it is the most
reliable current way:
- only one type of address avoids lots of complications which are not
relevant for these tests.
- there are contexts where IPv6 is not available, whereas those where
the IPv4 loopback is not are excessively rare.
"""
port = attr.ib(default=None)
sshd_exe_path = attr.ib(default='/usr/sbin/sshd')
initialized = False
def init(self):
self.user_name = pwd.getpwuid(os.getuid()).pw_name
self.init_address()
self.host = self.host_ipv4
self.init_config()
self.initialized = True
def init_address(self): # pragma no cover (depends on tests invocation)
"""Determine the port for the server to bind to.
For best reliability in controlled environment such as CI,
the port is read from the `HGITALY_TESTS_SSHD_PORT` environment
variable.
Otherwise, for developer convenience, we determine an available port.
This is inherently racy, as the port may well not be available any
more when the SSH server actually starts.
"""
from_env = os.environ.get('HGITALY_TESTS_SSHD_PORT')
if from_env is not None:
self.port = int(from_env)
else:
sock = socket.create_server((self.host_ipv4, 0),
family=socket.AF_INET)
self.port = sock.getsockname()[1]
sock.close()
def init_config(self):
config_path = self.sshd_config_path = self.working_dir / 'sshd_config'
config_path.write_text(
self.config_template_path.read_text().format(
user=self.user_name,
host=self.host,
port=self.port,
authorized_keys_path=self.authorized_keys_path),
encoding='ascii')
perms = 0o600 # not tracked by hg, won't look as a file modification
self.host_pub_key_path.chmod(perms)
self.host_priv_key_path.chmod(perms)
def start(self):
self.process = subprocess.Popen((self.sshd_exe_path,
'-h', self.host_priv_key_path,
'-c', self.host_pub_key_path,
'-f', self.sshd_config_path,
'-D', '-e',
),
)
def stop(self):
self.process.terminate()
self.process.wait()
def hg_ssh_setup(working_dir):
server = SSHServer(
host_pub_key_path=TEST_DATA_DIR / 'ssh_host_ecdsa_key.pub',
host_priv_key_path=TEST_DATA_DIR / 'ssh_host_ecdsa_key',
config_template_path=TEST_DATA_DIR / 'sshd_config',
authorized_keys_path=TEST_DATA_DIR / 'authorized_keys',
working_dir=working_dir,
)
server.init()
client_key = (TEST_DATA_DIR / 'id_ecdsa_user').read_text()
known_hosts = (TEST_DATA_DIR / 'known_hosts').read_text().format(
host=server.host, port=server.port)
server.start()
yield server, client_key, known_hosts
server.stop()
# Copyright 2021 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
#
# SPDX-License-Identifier: GPL-2.0-or-later
# Copyright 2021 Georges Racinet <georges.racinet@octobus.net>
#
# This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version.
#
# SPDX-License-Identifier: GPL-2.0-or-later
import pytest
from mercurial_testhelpers.repo_wrapper import RepoWrapper
from ..sshd import hg_ssh_setup
from hgitaly.ssh import client_command_str
from hgitaly.testing.ssh import hg_exe_path
@pytest.fixture
def ssh_setup(tmpdir):
working_dir = tmpdir / 'working'
working_dir.mkdir()
yield from hg_ssh_setup(working_dir)
def test_ssh_server(ssh_setup, tmpdir):
server, client_key, known_hosts = ssh_setup
src = RepoWrapper.init(tmpdir / 'src')
dest = RepoWrapper.init(tmpdir / 'dst')
ctx = src.commit_file('foo')
with client_command_str(client_key, known_hosts) as ssh_cmd:
src.command('push',
f'ssh://{server.host}:{server.port}/{dest.path}'.encode(),
ssh=ssh_cmd.encode(),
remotecmd=hg_exe_path().encode(),
)
assert ctx in dest.repo
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment