From 998b774d5f0fddf6b113da8102b64680c0ece43c Mon Sep 17 00:00:00 2001 From: Jakub Jelen Date: Thu, 25 Apr 2024 17:16:32 +0200 Subject: [PATCH] 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 Co-Authored-By: Michal Hecko do not use filesystem during tests --- .../libraries/readopensshconfig.py | 61 ++++-- ..._readopensshconfig_opensshconfigscanner.py | 180 +++++++++++++++++- 2 files changed, 229 insertions(+), 12 deletions(-) diff --git a/repos/system_upgrade/common/actors/opensshconfigscanner/libraries/readopensshconfig.py b/repos/system_upgrade/common/actors/opensshconfigscanner/libraries/readopensshconfig.py index e6cb9fcc1e..50e37092d7 100644 --- a/repos/system_upgrade/common/actors/opensshconfigscanner/libraries/readopensshconfig.py +++ b/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 diff --git a/repos/system_upgrade/common/actors/opensshconfigscanner/tests/test_readopensshconfig_opensshconfigscanner.py b/repos/system_upgrade/common/actors/opensshconfigscanner/tests/test_readopensshconfig_opensshconfigscanner.py index 68a9ec47ec..64c16f7fd1 100644 --- a/repos/system_upgrade/common/actors/opensshconfigscanner/tests/test_readopensshconfig_opensshconfigscanner.py +++ b/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 = []