OpenSSHConfigScanner: Include directive is supported since RHEL 8.6

This issue could cause false positive reports when the user has the
configuration options such as "Subsystem sftp" defined in included file
only.

Resolves: RHEL-33902

Signed-off-by: Jakub Jelen <[email protected]>
Co-Authored-By: Michal Hecko <[email protected]>

do not use filesystem during tests

Author: Jakuje

repos/system_upgrade/common/actors/opensshconfigscanner/libraries/readopensshconfig.py

@@ -1,5 +1,9 @@
 import errno
+import glob
+import os
+import shlex
 
+from leapp.exceptions import StopActorExecutionError
 from leapp.libraries.common.rpms import check_file_modification
 from leapp.libraries.stdlib import api
 from leapp.models import OpenSshConfig, OpenSshPermitRootLogin
@@ -12,14 +16,31 @@ def line_empty(line):
     return len(line) == 0 or line.startswith('\n') or line.startswith('#')
 
 
-def parse_config(config):
-    """Parse OpenSSH server configuration or the output of sshd test option."""
+def parse_config(config, base_config=None, current_cfg_depth=0):
+    """
+    Parse OpenSSH server configuration or the output of sshd test option.
 
-    # RHEL7 defaults
-    ret = OpenSshConfig(
-        permit_root_login=[],
-        deprecated_directives=[]
-    )
+    :param Optional[OpenSshConfig] base_config: Base configuration that is extended with configuration options from
+    current file.
+
+    :param int current_cfg_depth: Internal counter for how many includes were already followed.
+    """
+
+    if current_cfg_depth > 16:
+        # This should really never happen as it would mean the SSH server won't
+        # start anyway on the old system.
+        error = 'Too many recursive includes while parsing sshd_config'
+        api.current_logger().error(error)
+        return None
+
+    ret = base_config
+    if ret is None:
+        # RHEL7 defaults
+        ret = OpenSshConfig(
+            permit_root_login=[],
+            deprecated_directives=[]
+        )
+        # TODO(Jakuje): Do we need different defaults for RHEL8?
 
     in_match = None
     for line in config:
@@ -68,8 +89,26 @@ def parse_config(config):
                 # here we need to record all remaining items as command and arguments
                 ret.subsystem_sftp = ' '.join(el[2:])
 
+        elif el[0].lower() == 'include':
+            # recursively parse the given file or files referenced by this option
+            # the option can have several space-separated filenames with glob wildcards
+            for pattern in shlex.split(' '.join(el[1:])):
+                if pattern[0] != '/' and pattern[0] != '~':
+                    pattern = os.path.join('/etc/ssh/', pattern)
+                files_matching_include_pattern = glob.glob(pattern)
+                # OpenSSH sorts the files lexicographically
+                files_matching_include_pattern.sort()
+                for included_config_file in files_matching_include_pattern:
+                    output = read_sshd_config(included_config_file)
+                    if parse_config(output, base_config=ret, current_cfg_depth=current_cfg_depth + 1) is None:
+                        raise StopActorExecutionError(
+                            'Failed to parse sshd configuration file',
+                            details={'details': 'Too many recursive includes while parsing {}.'
+                                     .format(included_config_file)}
+                        )
+
         elif el[0].lower() in DEPRECATED_DIRECTIVES:
-            # Filter out duplicit occurrences of the same deprecated directive
+            # Filter out duplicate occurrences of the same deprecated directive
             if el[0].lower() not in ret.deprecated_directives:
                 # Use the directive in the form as found in config for user convenience
                 ret.deprecated_directives.append(el[0])
@@ -82,10 +121,10 @@ def produce_config(producer, config):
     producer(config)
 
 
-def read_sshd_config():
+def read_sshd_config(config):
     """Read the actual sshd configuration file."""
     try:
-        with open(CONFIG, 'r') as fd:
+        with open(config, 'r') as fd:
             return fd.readlines()
     except IOError as err:
         if err.errno != errno.ENOENT:
@@ -98,7 +137,7 @@ def scan_sshd(producer):
     """Parse sshd_config configuration file to create OpenSshConfig message."""
 
     # direct access to configuration file
-    output = read_sshd_config()
+    output = read_sshd_config(CONFIG)
     config = parse_config(output)
 
     # find out whether the file was modified from the one shipped in rpm

repos/system_upgrade/common/actors/opensshconfigscanner/tests/test_readopensshconfig_opensshconfigscanner.py

@@ -1,3 +1,12 @@
+import glob
+import os
+import shutil
+import tempfile
+
+import pytest
+
+from leapp.exceptions import StopActorExecutionError
+from leapp.libraries.actor import readopensshconfig
 from leapp.libraries.actor.readopensshconfig import line_empty, parse_config, produce_config
 from leapp.models import OpenSshConfig, OpenSshPermitRootLogin
 
@@ -143,12 +152,181 @@ def test_parse_config_deprecated():
 def test_parse_config_empty():
     output = parse_config([])
     assert isinstance(output, OpenSshConfig)
-    assert isinstance(output, OpenSshConfig)
     assert not output.permit_root_login
     assert output.use_privilege_separation is None
     assert output.protocol is None
 
 
+def test_parse_config_include(monkeypatch):
+    """ This already require some files to touch """
+
+    config_contents = {
+        '/etc/ssh/sshd_config': [
+            "Include /path/*.conf"
+        ],
+        '/path/my.conf': [
+            'Subsystem sftp internal-sftp'
+        ],
+        '/path/another.conf': [
+            'permitrootlogin no'
+        ]
+    }
+
+    primary_config_path = '/etc/ssh/sshd_config'
+    primary_config_contents = config_contents[primary_config_path]
+
+    def glob_mocked(pattern):
+        assert pattern == '/path/*.conf'
+        return ['/path/my.conf', '/path/another.conf']
+
+    def read_config_mocked(path):
+        return config_contents[path]
+
+    monkeypatch.setattr(glob, 'glob', glob_mocked)
+    monkeypatch.setattr(readopensshconfig, 'read_sshd_config', read_config_mocked)
+
+    output = parse_config(primary_config_contents)
+
+    assert isinstance(output, OpenSshConfig)
+    assert len(output.permit_root_login) == 1
+    assert output.permit_root_login[0].value == 'no'
+    assert output.permit_root_login[0].in_match is None
+    assert output.use_privilege_separation is None
+    assert output.protocol is None
+    assert output.subsystem_sftp == 'internal-sftp'
+
+
+def test_parse_config_include_recursive(monkeypatch):
+    """ The recursive include should gracefully fail """
+
+    config_contents = {
+        '/etc/ssh/sshd_config': [
+            "Include /path/*.conf"
+        ],
+        '/path/recursive.conf': [
+            "Include /path/*.conf"
+        ],
+    }
+
+    primary_config_path = '/etc/ssh/sshd_config'
+    primary_config_contents = config_contents[primary_config_path]
+
+    def glob_mocked(pattern):
+        assert pattern == '/path/*.conf'
+        return ['/path/recursive.conf']
+
+    def read_config_mocked(path):
+        return config_contents[path]
+
+    monkeypatch.setattr(glob, 'glob', glob_mocked)
+    monkeypatch.setattr(readopensshconfig, 'read_sshd_config', read_config_mocked)
+
+    with pytest.raises(StopActorExecutionError) as recursive_error:
+        parse_config(primary_config_contents)
+    assert 'Failed to parse sshd configuration file' in str(recursive_error)
+
+
+def test_parse_config_include_relative(monkeypatch):
+    """ When the include argument is relative path, it should point into the /etc/ssh/ """
+
+    config_contents = {
+        '/etc/ssh/sshd_config': [
+            "Include relative/*.conf"
+        ],
+        '/etc/ssh/relative/default.conf': [
+            'Match address 192.168.1.42',
+            'PermitRootLogin yes'
+        ],
+        '/etc/ssh/relative/other.conf': [
+            'Match all',
+            'PermitRootLogin prohibit-password'
+        ],
+        '/etc/ssh/relative/wrong.extension': [
+            "macs hmac-md5",
+        ],
+    }
+
+    primary_config_path = '/etc/ssh/sshd_config'
+    primary_config_contents = config_contents[primary_config_path]
+
+    def glob_mocked(pattern):
+        assert pattern == '/etc/ssh/relative/*.conf'
+        return ['/etc/ssh/relative/other.conf', '/etc/ssh/relative/default.conf']
+
+    def read_config_mocked(path):
+        return config_contents[path]
+
+    monkeypatch.setattr(glob, 'glob', glob_mocked)
+    monkeypatch.setattr(readopensshconfig, 'read_sshd_config', read_config_mocked)
+
+    output = parse_config(primary_config_contents)
+
+    assert isinstance(output, OpenSshConfig)
+    assert len(output.permit_root_login) == 2
+    assert output.permit_root_login[0].value == 'yes'
+    assert output.permit_root_login[0].in_match == ['address', '192.168.1.42']
+    assert output.permit_root_login[1].value == 'prohibit-password'
+    assert output.permit_root_login[1].in_match == ['all']
+    assert output.use_privilege_separation is None
+    assert output.ciphers is None
+    assert output.macs is None
+    assert output.protocol is None
+    assert output.subsystem_sftp is None
+
+
+def test_parse_config_include_complex(monkeypatch):
+    """ This already require some files to touch """
+
+    config_contents = {
+        '/etc/ssh/sshd_config': [
+            "Include /path/*.conf /other/path/*.conf \"/last/path with spaces/*.conf\" "
+        ],
+        '/path/my.conf': [
+            'permitrootlogin prohibit-password'
+        ],
+        '/other/path/another.conf': [
+            'ciphers aes128-ctr'
+        ],
+        '/last/path with spaces/filename with spaces.conf': [
+            'subsystem sftp other-internal'
+        ]
+    }
+    glob_contents = {
+        '/path/*.conf': [
+            '/path/my.conf'
+        ],
+        '/other/path/*.conf': [
+            '/other/path/another.conf'
+        ],
+        '/last/path with spaces/*.conf': [
+            '/last/path with spaces/filename with spaces.conf'
+        ],
+    }
+
+    primary_config_path = '/etc/ssh/sshd_config'
+    primary_config_contents = config_contents[primary_config_path]
+
+    def glob_mocked(pattern):
+        return glob_contents[pattern]
+
+    def read_config_mocked(path):
+        return config_contents[path]
+
+    monkeypatch.setattr(glob, 'glob', glob_mocked)
+    monkeypatch.setattr(readopensshconfig, 'read_sshd_config', read_config_mocked)
+
+    output = parse_config(primary_config_contents)
+
+    assert isinstance(output, OpenSshConfig)
+    assert len(output.permit_root_login) == 1
+    assert output.permit_root_login[0].value == 'prohibit-password'
+    assert output.permit_root_login[0].in_match is None
+    assert output.use_privilege_separation is None
+    assert output.ciphers == "aes128-ctr"
+    assert output.protocol is None
+    assert output.subsystem_sftp == 'other-internal'
+
+
 def test_produce_config():
     output = []