mgr/cephadm: validate user provided ssh_config

ensure the user provided ssh_config contains a valid
`StrictHostKeyChecking` setting

Fixes: https://tracker.ceph.com/issues/48158
Signed-off-by: Michael Fritch <mfritch@suse.com>

diff --git a/src/pybind/mgr/cephadm/module.py b/src/pybind/mgr/cephadm/module.py
index fe10fbb74cbb5b790ebc095143528cdb8d11e3d3..976cfd2e81171dc4539ef1434f166828b7e2f13c 100644 (file)
--- a/src/pybind/mgr/cephadm/module.py
+++ b/src/pybind/mgr/cephadm/module.py
@@ -1,6 +1,7 @@
 import json
 import errno
 import logging
+import re
 import shlex
 from collections import defaultdict
 from configparser import ConfigParser
@@ -598,6 +599,17 @@ class CephadmOrchestrator(orchestrator.Orchestrator, MgrModule,
 
         self._reset_cons()
 
+    def validate_ssh_config_content(self, ssh_config):
+        if ssh_config is None or len(ssh_config.strip()) == 0:
+            raise OrchestratorValidationError('ssh_config cannot be empty')
+        # StrictHostKeyChecking is [yes|no] ?
+        l = re.findall(r'StrictHostKeyChecking\s+.*', ssh_config)
+        if not l:
+            raise OrchestratorValidationError('ssh_config requires StrictHostKeyChecking')
+        for s in l:
+            if 'ask' in s.lower():
+                raise OrchestratorValidationError(f'ssh_config cannot contain: \'{s}\'')
+
     def validate_ssh_config_fname(self, ssh_config_fname):
         if not os.path.isfile(ssh_config_fname):
             raise OrchestratorValidationError("ssh_config \"{}\" does not exist".format(
@@ -657,14 +669,10 @@ class CephadmOrchestrator(orchestrator.Orchestrator, MgrModule,
     def _set_ssh_config(self, inbuf=None):
         """
         Set an ssh_config file provided from stdin
-
-        TODO:
-          - validation
         """
-        if inbuf is None or len(inbuf) == 0:
-            return -errno.EINVAL, "", "empty ssh config provided"
         if inbuf == self.ssh_config:
             return 0, "value unchanged", ""
+        self.validate_ssh_config_content(inbuf)
         self.set_store("ssh_config", inbuf)
         self.log.info('Set ssh_config')
         self._reconfig_ssh()