[fix] Make the SSH missing host key policy configurable

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
+<https://docs.paramiko.org/en/stable/api/client.html#paramiko.client.MissingHostKeyPolicy>`_
+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``
 -----------------------------
 

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):

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)

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)