Method evaluate

src/ssh_audit/policy.py:334–464

Evaluates a server configuration against this policy. Returns a tuple of a boolean (True if server adheres to policy) and an array of strings that holds error messages.

(self, banner: Optional['Banner'], kex: Optional['SSH2_Kex'])

    def evaluate(self, banner: Optional['Banner'], kex: Optional['SSH2_Kex']) -> Tuple[bool, List[Dict[str, str]], str]:
        '''Evaluates a server configuration against this policy.  Returns a tuple of a boolean (True if server adheres to policy) and an array of strings that holds error messages.'''

        ret = True

        banner_str = str(banner)
        if (self._banner is not None) and (banner_str != self._banner):
            ret = False
            self._append_error('Banner', [self._banner], None, [banner_str])

        # All subsequent tests require a valid kex, so end here if we don't have one.
        if kex is None:
            error_list, error_str = self._get_errors()
            return ret, error_list, error_str

        if (self._compressions is not None) and (kex.server.compression != self._compressions):
            ret = False
            self._append_error('Compression', self._compressions, None, kex.server.compression)

        # If a list of optional host keys was given in the policy, remove any of its entries from the list retrieved from the server.  This allows us to do an exact comparison with the expected list below.
        pruned_host_keys = kex.key_algorithms
        if self._optional_host_keys is not None:
            pruned_host_keys = [x for x in kex.key_algorithms if x not in self._optional_host_keys]

        # Check host keys.
        if self._host_keys is not None:
            # If the policy allows subsets and re-ordered algorithms...
            if self._allow_algorithm_subset_and_reordering:
                for hostkey_t in kex.key_algorithms:
                    if hostkey_t not in self._host_keys:
                        ret = False
                        self._append_error('Host keys', self._host_keys, self._optional_host_keys, kex.key_algorithms)
                        break
            # The policy requires exact matching of algorithms.
            elif pruned_host_keys != self._host_keys:
                ret = False
                self._append_error('Host keys', self._host_keys, self._optional_host_keys, kex.key_algorithms)

        # Check host key sizes.
        if self._hostkey_sizes is not None:
            hostkey_types = list(self._hostkey_sizes.keys())
            hostkey_types.sort()  # Sorted to make testing output repeatable.
            for hostkey_type in hostkey_types:
                expected_hostkey_size = cast(int, self._hostkey_sizes[hostkey_type]['hostkey_size'])
                server_host_keys = kex.host_keys()
                if hostkey_type in server_host_keys:
                    actual_hostkey_size = cast(int, server_host_keys[hostkey_type]['hostkey_size'])
                    if (self._allow_larger_keys and actual_hostkey_size < expected_hostkey_size) or \
                       (not self._allow_larger_keys and actual_hostkey_size != expected_hostkey_size):
                        ret = False
                        self._append_error('Host key (%s) sizes' % hostkey_type, [str(expected_hostkey_size)], None, [str(actual_hostkey_size)])

                    # If we have expected CA signatures set, check them against what the server returned.
                    if self._hostkey_sizes is not None and len(cast(str, self._hostkey_sizes[hostkey_type]['ca_key_type'])) > 0 and cast(int, self._hostkey_sizes[hostkey_type]['ca_key_size']) > 0:
                        expected_ca_key_type = cast(str, self._hostkey_sizes[hostkey_type]['ca_key_type'])
                        expected_ca_key_size = cast(int, self._hostkey_sizes[hostkey_type]['ca_key_size'])
                        actual_ca_key_type = cast(str, server_host_keys[hostkey_type]['ca_key_type'])
                        actual_ca_key_size = cast(int, server_host_keys[hostkey_type]['ca_key_size'])