diff --git a/docs/user/settings.rst b/docs/user/settings.rst index aa59425ef..6449ab25c 100644 --- a/docs/user/settings.rst +++ b/docs/user/settings.rst @@ -3,6 +3,26 @@ Settings .. include:: /partials/settings-note.rst +``OPENWISP_SSH_MISSING_HOST_KEY_POLICY`` +---------------------------------------- + +============ ========================== +**type**: ``str`` +**default**: ``paramiko.AutoAddPolicy`` +============ ========================== + +Dotted path to the `paramiko missing host key policy +`_ +used when establishing SSH connections to devices. + +The default ``paramiko.AutoAddPolicy`` accepts unknown device host keys +automatically, which is convenient but leaves the configuration push (and +the device secrets it carries) exposed to a man-in-the-middle on the +management network. Set this to a stricter policy (for example +``paramiko.RejectPolicy``) to enforce host key verification; in that case +the system known hosts are loaded so that trusted device host keys can +still be verified. + ``OPENWISP_SSH_AUTH_TIMEOUT`` ----------------------------- diff --git a/openwisp_controller/connection/connectors/ssh.py b/openwisp_controller/connection/connectors/ssh.py index 52a585304..a655f9913 100644 --- a/openwisp_controller/connection/connectors/ssh.py +++ b/openwisp_controller/connection/connectors/ssh.py @@ -5,6 +5,7 @@ import paramiko from django.utils.functional import cached_property +from django.utils.module_loading import import_string from django.utils.translation import gettext_lazy as _ from jsonschema import validate from jsonschema.exceptions import ValidationError as SchemaError @@ -61,7 +62,20 @@ def __init__(self, params, addresses): self._params = params self.addresses = addresses self.shell = paramiko.SSHClient() - self.shell.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + self.shell.set_missing_host_key_policy(self._get_missing_host_key_policy()) + + def _get_missing_host_key_policy(self): + # The default AutoAddPolicy blindly trusts unknown host keys, which + # exposes the configuration push -- and the device secrets it carries + # (WiFi passwords, VPN private keys, ...) -- to a man-in-the-middle on + # the management path. Operators can enforce host-key verification by + # setting OPENWISP_SSH_MISSING_HOST_KEY_POLICY to a stricter policy + # (e.g. "paramiko.RejectPolicy"); in that case the known host keys are + # loaded so trusted devices can still be verified. + policy_class = import_string(app_settings.SSH_MISSING_HOST_KEY_POLICY) + if not issubclass(policy_class, paramiko.AutoAddPolicy): + self.shell.load_system_host_keys() + return policy_class() @classmethod def validate(cls, params): diff --git a/openwisp_controller/connection/settings.py b/openwisp_controller/connection/settings.py index 50223a137..722326aa2 100644 --- a/openwisp_controller/connection/settings.py +++ b/openwisp_controller/connection/settings.py @@ -35,6 +35,11 @@ }, ) +SSH_MISSING_HOST_KEY_POLICY = getattr( + settings, + "OPENWISP_SSH_MISSING_HOST_KEY_POLICY", + "paramiko.AutoAddPolicy", +) SSH_AUTH_TIMEOUT = getattr(settings, "OPENWISP_SSH_AUTH_TIMEOUT", 2) SSH_BANNER_TIMEOUT = getattr(settings, "OPENWISP_SSH_BANNER_TIMEOUT", 60) SSH_COMMAND_TIMEOUT = getattr(settings, "OPENWISP_SSH_COMMAND_TIMEOUT", 30) diff --git a/openwisp_controller/connection/tests/test_ssh.py b/openwisp_controller/connection/tests/test_ssh.py index fa182d424..55797d209 100644 --- a/openwisp_controller/connection/tests/test_ssh.py +++ b/openwisp_controller/connection/tests/test_ssh.py @@ -1,11 +1,14 @@ import os from unittest import mock +import paramiko from django.conf import settings from django.test import TestCase from paramiko.ssh_exception import AuthenticationException from swapper import load_model +from .. import settings as app_settings +from ..connectors.ssh import Ssh from ..connectors.ssh import logger as ssh_logger from .utils import CreateConnectionsMixin, SshServer @@ -29,6 +32,25 @@ def tearDownClass(cls): super().tearDownClass() cls.mock_ssh_server.__exit__() + def test_default_missing_host_key_policy(self): + connector = Ssh( + params={"username": "root", "password": "password"}, + addresses=["127.0.0.1"], + ) + self.assertIsInstance(connector.shell._policy, paramiko.AutoAddPolicy) + + @mock.patch.object( + app_settings, "SSH_MISSING_HOST_KEY_POLICY", "paramiko.RejectPolicy" + ) + @mock.patch("paramiko.SSHClient.load_system_host_keys") + def test_configurable_missing_host_key_policy(self, load_system_host_keys): + connector = Ssh( + params={"username": "root", "password": "password"}, + addresses=["127.0.0.1"], + ) + self.assertIsInstance(connector.shell._policy, paramiko.RejectPolicy) + load_system_host_keys.assert_called_once() + @mock.patch.object(ssh_logger, "debug") def test_connection_connect(self, mocked_debug): ckey = self._create_credentials_with_key(port=self.ssh_server.port)