Merge pull request #288 from GreatFruitOmsk/277-cis-openssh

OpenSSH audit: range tests.

agent/__init__.py

@@ -834,19 +834,6 @@ def run(ping=True, dev=False, logger=logger):
         os.chmod(INI_PATH, 0o600)
 
 
-def patch(name):
-    openssh_params = {
-        'openssh-empty-password': 'PermitEmptyPasswords',
-        'openssh-root-login': 'PermitRootLogin',
-        'openssh-password-auth': 'PasswordAuthentication',
-        'openssh-agent-forwarding': 'AllowAgentForwarding',
-        'openssh-protocol': 'Protocol'
-    }
-    param = openssh_params[name]
-    logger.info('patch "{}"'.format(param))
-    security_helper.patch_sshd_config(param)
-
-
 def upgrade(packages):
     logger.info('upgrade packages: {}'.format(packages))
     upgrade_packages(packages)

agent/__main__.py

@@ -2,8 +2,9 @@
 import asyncio
 import logging
 
-from . import run, get_device_id, get_open_ports, say_hello, get_claim_token, get_claim_url, patch, upgrade, executor
+from . import run, get_device_id, get_open_ports, say_hello, get_claim_token, get_claim_url, upgrade, executor
 from . import fetch_credentials, fetch_device_metadata, setup_logging
+from .security_helper import patch_sshd_config
 
 logger = logging.getLogger('agent')
 
@@ -21,14 +22,37 @@ def main():
     }
 
     patches = {
-        'openssh-empty-password': 'OpenSSH: Disable logins with empty password',
-        'openssh-root-login': 'OpenSSH: Disable root login',
-        'openssh-password-auth': 'OpenSSH: Disable password authentication',
-        'openssh-agent-forwarding': 'OpenSSH: Disable agent forwarding',
-        'openssh-protocol': '\tOpenSSH: Force protocol version 2'
+        'openssh-empty-password':
+            ('OpenSSH: Disable logins with empty password', 'PermitEmptyPasswords'),
+        'openssh-root-login':
+            ('OpenSSH: Disable root login', 'PermitRootLogin'),
+        'openssh-password-auth':
+            ('OpenSSH: Disable password authentication', 'PasswordAuthentication'),
+        'openssh-agent-forwarding':
+            ('OpenSSH: Disable agent forwarding', 'AllowAgentForwarding'),
+        'openssh-protocol':
+            ('\tOpenSSH: Force protocol version 2', 'Protocol'),
+        'openssh-client-alive-interval':
+            ('OpenSSH: Active Client Interval', 'ClientAliveInterval'),
+        'openssh-client-alive-count-max':
+            ('OpenSSH: Active Client Max Count', 'ClientAliveCountMax'),
+        'openssh-host-based-auth':
+            ('OpenSSH: Host-based Authentication', 'HostbasedAuthentication'),
+        'openssh-ignore-rhosts':
+            ('OpenSSH: Ignore rhosts', 'IgnoreRhosts'),
+        'openssh-log-level':
+            ('\tOpenSSH: Log Level', 'LogLevel'),
+        'openssh-login-grace-time':
+            ('OpenSSH: Login Grace Time', 'LoginGraceTime'),
+        'openssh-max-auth-tries':
+            ('OpenSSH: Max Auth Tries', 'MaxAuthTries'),
+        'openssh-permit-user-env':
+            ('OpenSSH: Permit User Environment', 'PermitUserEnvironment'),
+        'openssh-x11-forwarding':
+            ('OpenSSH: X11 Forwarding', 'X11Forwarding')
     }
     patch_help_string = "One of the following:\n" + "\n".join(
-        ["{}\t{}".format(k, v) for k, v in patches.items()])
+        ["{}\t{}".format(k, v[0]) for k, v in patches.items()])
 
     parser = argparse.ArgumentParser(
         formatter_class=argparse.RawTextHelpFormatter,
@@ -78,7 +102,7 @@ def main():
         logger.info("start in daemon mode...")
         run_daemon(dev=args.dev)
     elif action == 'patch':
-        patch(args.patch_name)
+        patch_sshd_config(patches[args.patch_name][1])
         run(ping=True, dev=args.dev)
     elif action == 'upgrade':
         upgrade(args.packages)

agent/security_helper.py

@@ -6,6 +6,7 @@
 import socket
 import subprocess
 import time
+from enum import IntEnum
 from hashlib import sha256
 from pathlib import Path
 from socket import SocketKind
@@ -163,19 +164,63 @@ def selinux_status():
     '/etc/shadow',
     '/etc/group'
 ]
-
+BLOCK_SIZE = 64 * 1024
 SSHD_CONFIG_PATH = '/etc/ssh/sshd_config'
-# Value: (default, safe).
+
+
+class SshdConfigParam:
+    """
+    Encapsulates safe and default values for an OpenSSH parameter.
+    """
+    class COMPARE(IntEnum):
+        """
+        Supported comparsions.
+        MATCH: exact match
+        RANGE: inclusive integer range (min, max)
+        """
+        MATCH = 1
+        RANGE = 2
+
+    def _match(self, val: str) -> bool:
+        return self._safe == val
+
+    def _range(self, val: str) -> bool:
+        vmin, vmax = self._safe
+        return vmin <= int(val) <= vmax
+
+    @property
+    def safe_value(self) -> str:
+        """
+        :return: safe value
+        """
+        return self._safe_value
+
+    def __init__(self, default, safe, compare=COMPARE.MATCH):
+        self.default = default
+        self._safe = safe
+        # select compare function which decides if the value is safe.
+        self.is_safe = {self.COMPARE.MATCH: self._match,
+                        self.COMPARE.RANGE: self._range}[compare]
+        self._safe_value = safe if compare == self.COMPARE.MATCH else str(self._safe[1])
+
+
 SSHD_CONFIG_PARAMS_INFO = {
-    'PermitEmptyPasswords': ['no', 'no'],
-    'PermitRootLogin': ['yes', 'no'],
-    'PasswordAuthentication': ['yes', 'no'],
-    'AllowAgentForwarding': ['yes', 'no'],
-    'Protocol': ['2', '2']
+    'PermitEmptyPasswords': SshdConfigParam('no', 'no'),
+    'PermitRootLogin': SshdConfigParam('yes', 'no'),
+    'PasswordAuthentication': SshdConfigParam('yes', 'no'),
+    'AllowAgentForwarding': SshdConfigParam('yes', 'no'),
+    'Protocol': SshdConfigParam('2', '2'),
+    'ClientAliveInterval': SshdConfigParam('0', (1, 300), SshdConfigParam.COMPARE.RANGE),
+    'ClientAliveCountMax': SshdConfigParam('3', (0, 3), SshdConfigParam.COMPARE.RANGE),
+    'HostbasedAuthentication': SshdConfigParam('no', 'no'),
+    'IgnoreRhosts': SshdConfigParam('yes', 'yes'),
+    'LogLevel': SshdConfigParam('INFO', 'INFO'),
+    'LoginGraceTime': SshdConfigParam('120', (1, 60), SshdConfigParam.COMPARE.RANGE),
+    'MaxAuthTries': SshdConfigParam('6', (0, 4), SshdConfigParam.COMPARE.RANGE),
+    'PermitUserEnvironment': SshdConfigParam('no', 'no'),
+    'X11Forwarding': SshdConfigParam('no', 'no')
 }
 
-BLOCK_SIZE = 64 * 1024
-
 
 def audit_config_files():
     """
@@ -234,10 +279,10 @@ def audit_sshd():
     if sshd_version is not None and sshd_version >= 7.0:
         # According to https://www.openssh.com/releasenotes.html those things were changed in 7.0.
         del (config['Protocol'])
-        config['PermitRootLogin'][0] = 'prohibit-password'
+        config['PermitRootLogin'].default = 'prohibit-password'
 
-    # Fill the dict with default values which are gonna be updated with found config parameters' values.
-    insecure_params = {k: config[k][0] for k in config}
+    # Fill the dict with default values which will be updated with found config parameters' values.
+    insecure_params = {k: config[k].default for k in config}
     with open(SSHD_CONFIG_PATH) as sshd_config:
         for line in sshd_config:
             line = line.strip()
@@ -256,7 +301,7 @@ def audit_sshd():
                 insecure_params[parameter] = value
     issues = {}
     for param in insecure_params:
-        if insecure_params[param] != config[param][1]:
+        if not config[param].is_safe(insecure_params[param]):
             issues[param] = insecure_params[param]
     return issues
 
@@ -352,17 +397,18 @@ def cpu_vulnerabilities():
 def patch_sshd_config(patch_param):
     from . import BACKUPS_PATH
 
-    default_value, safe_value = SSHD_CONFIG_PARAMS_INFO[patch_param]
+    param_info = SSHD_CONFIG_PARAMS_INFO[patch_param]
     if not os.path.isfile(SSHD_CONFIG_PATH):
-        logger.error('{} not found'.format(SSHD_CONFIG_PATH))
+        logger.error('%s not found', SSHD_CONFIG_PATH)
         return
     try:
         from sh import sshd, service
     except ImportError:
         logger.exception('sshd or service executable not found')
         return
 
-    safe_value_string = '\n# Added by wott-agent on {}\n{} {}\n'.format(time.ctime(), patch_param, safe_value)
+    safe_value_string = '\n# Added by wott-agent on {}\n{} {}\n'.format(
+        time.ctime(), patch_param, param_info.safe_value)
     backup_filename = os.path.join(BACKUPS_PATH, 'sshd_config.' + str(int(time.time())))
     replaced = False
 
@@ -382,15 +428,15 @@ def patch_sshd_config(patch_param):
             param, value = line_split
             value = value.strip('"')
             if param == patch_param:
-                if value != safe_value:
-                    logger.info('{}: replacing "{}" with "{}"'.format(param, value, safe_value))
+                if not param_info.is_safe(value):
+                    logger.info('%s: replacing "%s" with "%s"', param, value, param_info.safe_value)
                     patched_lines[-1] = safe_value_string
                     replaced = True
                     safe = False
                 else:
                     safe = True
-        if not replaced and not safe and default_value != safe_value:
-            logger.info('{}: replacing default "{}" with "{}"'.format(patch_param, default_value, safe_value))
+        if not replaced and not safe and not param_info.is_safe(param_info.default):
+            logger.info('%s: replacing default "%s" with "%s"', patch_param, param_info.default, param_info.safe_value)
             patched_lines.append(safe_value_string)
             replaced = True
         if replaced:
@@ -400,9 +446,9 @@ def patch_sshd_config(patch_param):
                         "and installed your SSH keys on this server. Failure to do so will result in that you "
                         "will be locked out. I have have my SSH key(s) installed:"):
                     return
-            logger.info('Backing up {} as {}'.format(SSHD_CONFIG_PATH, backup_filename))
+            logger.info('Backing up %s as %s', SSHD_CONFIG_PATH, backup_filename)
             shutil.copy(SSHD_CONFIG_PATH, backup_filename)
-            logger.info('Writing {}'.format(SSHD_CONFIG_PATH))
+            logger.info('Writing %s', SSHD_CONFIG_PATH)
             sshd_config.seek(0, 0)
             sshd_config.truncate()
             sshd_config.writelines(patched_lines)
@@ -414,7 +460,7 @@ def patch_sshd_config(patch_param):
         sshd('-t')
     except ErrorReturnCode_255 as e:
         if e.stderr.startswith(SSHD_CONFIG_PATH.encode()):
-            logger.exception('{} is invalid. Restoring from backup.'.format(SSHD_CONFIG_PATH))
+            logger.exception('%s is invalid. Restoring from backup.', SSHD_CONFIG_PATH)
             shutil.copy(backup_filename, SSHD_CONFIG_PATH)
         else:
             logger.exception('something went wrong')

tests/conftest.py

@@ -245,11 +245,21 @@ def sshd_config():
     return """
 # a comment
 PermitEmptyPasswords no
-
-# another comment
 PermitRootLogin   "yes"
+
+# Ignored with OpenSSH >= 7.0
 Protocol  "2,1"
-# PasswordAuthentication param's default value is gonna be checked
+
+# PasswordAuthentication param's default value will be checked
+LoginGraceTime 60
+
+# outside of range
+MaxAuthTries 5
+
+# inside the range
+ClientAliveCountMax 1
+
+# default: ClientAliveInterval 0
 
 AnotherOption another value
 """

tests/test_agent.py

@@ -423,7 +423,9 @@ def mock_open(filename, mode='r'):
             audit[3]['issues'] == {'PermitRootLogin': 'yes',
                                    'PasswordAuthentication': 'yes',
                                    'Protocol': '2,1',
-                                   'AllowAgentForwarding': 'yes'}
+                                   'AllowAgentForwarding': 'yes',
+                                   'ClientAliveInterval': '0',
+                                   'MaxAuthTries': '5'}
 
 
 def test_block_networks(ipt_networks, ipt_rules):