def ssl_protos(bad_protos):
    bad_protos = set(bad_protos)
    results = GroupTestResult()

    try:
        config = _read_nginx_config('/etc/nginx/nginx.conf')
    except (ParsingError, EnvironmentError):
        return TestResult(Result.FAIL, "could not parse nginx config")
    http = _get_section(config, 'http')

    # check the default set in context 'http'
    default_protos = (_get_parameters(http, 'ssl_protocols') or
                      ['TLSv1', 'TLSv1.1', 'TLSv1.2'])

    # check each server separately
    for server in _config_iter_servers(http):
        if not _server_enables_ssl(server):
            continue

        name = '/'.join(_get_parameters(server, 'server_name'))
        server_protos = (_get_parameters(server, 'ssl_protocols') or
                         default_protos)
        forbidden = list(set(server_protos) & bad_protos)
        if forbidden:
            res = TestResult(Result.FAIL,
                             "server uses banned protocols: %s" %
                             ",".join(forbidden))
        else:
            res = TestResult(Result.PASS, "")
        results.add_result("server %s" % name, res)

    return results

def config_permission(config):
    try:
        user = pwd.getpwnam(config['user'])
    except KeyError:
        return TestResult(Result.SKIP,
                          'Could not find user "%s"' % config['user'])

    try:
        group = grp.getgrnam(config['group'])
    except KeyError:
        return TestResult(Result.SKIP,
                          'Could not find group "%s"' % config['group'])

    result = GroupTestResult()
    files = ['nova.conf',
             'api-paste.ini',
             'policy.json',
             'rootwrap.conf',
             ]
    for f in files:
        path = os.path.join(config['dir'], f)
        result.add_result(path,
                          utils.validate_permissions(path, 0o640, user.pw_uid,
                                                     group.gr_gid))
    return result

def test_docker_privilege():
    logger.debug("Testing if the container is running in user namespace.")
    notes = "No Docker containers found or docker is not running."

    results = GroupTestResult()

    containers = _get_docker_container()

    testcmd = '{{ .Id }}: {{.HostConfig.Privileged }}'

    if not containers:
        return TestResult(Result.SKIP, notes)

    for container_id in containers:
        if container_id == '':
            pass
        else:
            check = "Checking container: " + str(container_id)
            test = subprocess.check_output(['docker',
                                            'inspect',
                                            '--format',
                                            testcmd,
                                            container_id])

            entry = test.split(':')

            if 'false' in entry:
                result = TestResult(Result.PASS)
            else:
                notes = ("Container " + str(container_id) + " is running with "
                         "privileged flags set to true.")
                result = TestResult(Result.FAIL, notes)
            results.add_result(check, result)
    return results

def test_proc_map_access():
    tests = {
        "Can read own /proc/$pid/maps file": {
            "function": _can_read_from_own,
            "expected": True
        },
        "Can read others process /proc/$pid/maps with same UID": {
            "function": _can_read_any_with_same_uid,
            "expected": True
        },
        "Can't read /proc/$pid/maps of other processes": {
            "function": _cant_read_others,
            "expected": False
        },
        "Can't read parents after privileges were dropped": {
            "function": _cant_read_parents_when_priv_dropped,
            "expected": False
        }
    }
    results = GroupTestResult()
    for t in tests:
        fn = tests[t]["function"]
        exp = tests[t]["expected"]
        act = Value(ctypes.c_bool)
        p = Process(target=fn, args=(act,))
        p.start()
        p.join()

        result = Result.PASS if exp == act.value else Result.FAIL
        results.add_result(t, TestResult(result))

    return results

def usb_authorization():
    open_hosts = []
    hosts = [dev for dev in os.listdir('/sys/bus/usb/devices') if
             dev.startswith('usb')]

    for host in hosts:
        auth_file = os.path.join('/sys/bus/usb/devices', host,
                                 'authorized_default')
        if not os.path.isfile(auth_file):
            continue

        with open(auth_file, 'r') as f:
            contents = f.read().strip()

        if contents != '0':
            open_hosts.append(host)

    if not hosts:
        return TestResult(Result.SKIP, "no USB hosts found")

    if not open_hosts:
        return TestResult(Result.PASS, "no open USB hosts")

    results = GroupTestResult()
    for host in open_hosts:
        results.add_result(host, TestResult(
            Result.FAIL, "USB host accepts all devices by default"))
    return results

def test_ulimit_default_override():
    logger.debug("Testing if the container is running in user namespace.")
    notes = "No Docker containers found or docker is not running."

    results = GroupTestResult()

    containers = _get_docker_container()

    testcmd = '{{ .Id }}: Ulimits={{ .HostConfig.Ulimits }}'

    if not containers:
        return TestResult(Result.SKIP, notes)

    for container_id in containers:
        if container_id == '':
            pass
        else:
            check = "Checking container: " + str(container_id)
            test = subprocess.check_output(['docker',
                                            'inspect',
                                            '--format',
                                            testcmd,
                                            container_id])

            if '<no value>' in test:
                result = TestResult(Result.PASS)
            else:
                notes = ("Container " + str(container_id) + " is "
                         "running with default ulimits in place. ")
                result = TestResult(Result.FAIL, notes)

            results.add_result(check, result)
    return results

def config_permission(config):
    try:
        user = pwd.getpwnam(config['user'])
    except KeyError:
        return TestResult(Result.SKIP,
                          'Could not find user "%s"' % config['user'])

    try:
        group = grp.getgrnam(config['group'])
    except KeyError:
        return TestResult(Result.SKIP,
                          'Could not find group "%s"' % config['group'])

    result = GroupTestResult()
    files = ['keystone.conf',
             'keystone-paste.ini',
             'policy.json',
             'logging.conf',
             'ssl/certs/signing_cert.pem',
             'ssl/private/signing_key.pem',
             'ssl/certs/ca.pem',
             ]
    for f in files:
        path = os.path.join(config['dir'], f)
        result.add_result(path,
                          utils.validate_permissions(path, 0o640, user.pw_uid,
                                                     group.gr_gid))
    return result

def certificate_check(test_config):
    paths = glob.glob(test_config['configs'])
    if not paths:
        return TestResult(Result.SKIP, "No stunnel config found")

    results = GroupTestResult()

    for path in paths:
        config = _read_config(path)

        for section in config:
            cert_path = config[section].get('cert')
            # do this check only on sections with configured certificates
            if not cert_path:
                continue

            issues = utils.find_certificate_issues(cert_path)
            test_name = "%s:%s" % (path, section)
            if issues:
                msg = "problem in %s: %s" % (cert_path, issues)
                results.add_result(test_name, TestResult(Result.FAIL, msg))
            else:
                results.add_result(test_name, TestResult(Result.PASS))

    return results

def admin_token(config):
    try:
        path = os.path.join(config['dir'], 'keystone.conf')
        keystone_ini = utils.parse_openstack_ini(path)
        path = os.path.join(config['dir'], 'keystone-paste.ini')
        paste_ini = utils.parse_openstack_ini(path)
    except EnvironmentError:
        return TestResult(Result.SKIP, 'cannot read keystone config files')

    keystone_req = {
        "DEFAULT.admin_token": {"disallowed": "*"},
    }
    keystone_res = utils.verify_config("keystone.conf", keystone_ini,
                                       keystone_req, needs_parsing=False)

    paste_req = {
        "filter:admin_token_auth.AdminTokenAuthMiddleware": {"disallowed": "*"}
    }
    paste_res = utils.verify_config("keystone-paste.ini", paste_ini, paste_req,
                                    needs_parsing=False)

    result = GroupTestResult()
    for res in keystone_res:
        result.add_result(res[0], res[1])
    for res in paste_res:
        result.add_result(res[0], res[1])
    return result

def test_read_only_root_fs():
    logger.debug("Testing if the container is running in user namespace.")
    notes = "No Docker containers found or docker is not running."

    results = GroupTestResult()

    containers = _get_docker_container()

    testcmd = '{{ .Id }}: ReadonlyRootfs={{ .HostConfig.ReadonlyRootfs }}'

    if not containers:
        return TestResult(Result.SKIP, notes)

    for container_id in containers:
        if container_id == '':
            pass
        else:
            check = "Checking container: " + str(container_id)
            test = subprocess.check_output(['docker',
                                            'inspect',
                                            '--format',
                                            testcmd,
                                            container_id])

            if 'false' in test:
                result = TestResult(Result.PASS)
            else:
                notes = ("Container " + str(container_id) + " has a file "
                         "system with permissions that are not read only.")
                result = TestResult(Result.FAIL, notes)
            results.add_result(check, result)
    return results

def version_advertise():
    if not os.path.exists(NGINX_CONFIG_PATH):
        return TestResult(Result.SKIP, "nginx config not found")

    try:
        config = _read_nginx_config('/etc/nginx/nginx.conf')
    except (ParsingError, EnvironmentError):
        return TestResult(Result.FAIL, "could not parse nginx config")

    http = _get_section(config, 'http')
    results = GroupTestResult()

    default_tokens = _get_parameters(http, 'server_tokens') or 'on'

    for server in _config_iter_servers(http):
        name = '/'.join(_get_parameters(server, 'server_name'))
        tokens = _get_parameters(server, 'server_tokens')

        if (tokens or default_tokens) == 'on':
            res = TestResult(Result.FAIL,
                             "version is advertised (server_tokens)")
        else:
            res = TestResult(Result.PASS, "custom or hidden version")
        results.add_result("server %s" % name, res)

    return results

def test_sysctl_values(checks):
    results = GroupTestResult()

    if not checks:
        return TestResult(Result.SKIP, "Unable to load module config file")

    for key, pattern in checks.items():
        description = _sysctl_description(key, pattern)
        try:
            value = utils.get_sysctl_value(key)
            result = None
            if _sysctl_check(pattern, value):
                result = TestResult(Result.PASS)
            else:
                error = _sysctl_report_failure(pattern, value)
                result = TestResult(Result.FAIL, notes=error)

            results.add_result(description, result)

        except utils.ValNotFound:
            notes = "Could not find a value for {}".format(key)
            results.add_result(description,
                               TestResult(Result.SKIP, notes=notes))

    return results

def test_pax():
    pax_kernel_options = {
        "Non-executable kernel pages":          "CONFIG_PAX_KERNEXEC",
        "Non-executable pages":                 "CONFIG_PAX_NOEXEC",
        "Paging based non-executable pages":    "CONFIG_PAX_PAGEEXEC",
        "Restrict MPROTECT":                    "CONFIG_PAX_MPROTECT",
        "Address space layout randomization":   "CONFIG_PAX_ASLR",
        "Randomize kernel stack":               "CONFIG_PAX_RANDKSTACK",
        "Randomize user stack":                 "CONFIG_PAX_RANDUSTACK",
        "Randomize MMAP stack":                 "CONFIG_PAX_RANDMMAP",
        "Sanitize freed memory":                "CONFIG_PAX_MEMORY_SANITIZE",
        "Sanitize kernel stack":                "CONFIG_PAX_MEMORY_STACKLEAK",
        "Prevent userspace pointer deref":      "CONFIG_PAX_MEMORY_UDEREF",
        "Prevent kboject refcount overflow":    "CONFIG_PAX_REFCOUNT",
        "Bounds check heap object copies":      "CONFIG_PAX_USERCOPY",
    }

    config = utils.kconfig()
    if not config:
        return TestResult(Result.SKIP, notes="Unable to find kernel config")

    if not utils.kconfig_option('CONFIG_GRKERNSEC', config):
        return TestResult(Result.FAIL,
                          notes="Kernel not compiled with GRSECURITY patches")

    results = GroupTestResult()
    for test, setting in pax_kernel_options.items():
        enabled = utils.kconfig_option(setting, config)
        if enabled and enabled == 'y':
            results.add_result(test, TestResult(Result.PASS))
        else:
            results.add_result(test, TestResult(Result.FAIL))

    return results

def test_docker_pid_mode():
    logger.debug("Testing if the container is running in user namespace.")
    notes = "No Docker containers found or docker is not running."

    results = GroupTestResult()

    containers = _get_docker_container()

    testcmd = '{{ .Id }}: PidMode={{ .HostConfig.PidMode }}'

    if not containers:
        return TestResult(Result.SKIP, notes)

    for container_id in containers:
        if container_id == '':
            pass
        else:
            check = "Checking container: " + str(container_id)
            test = subprocess.check_output(['docker',
                                            'inspect',
                                            '--format',
                                            testcmd,
                                            container_id])

            if 'host' in test:
                notes = ("Container " + str(container_id) + " is sharing "
                         "host process namespaces.")
                result = TestResult(Result.FAIL, notes)
            else:
                result = TestResult(Result.PASS)

            results.add_result(check, result)
    return results

def test_mount_sensitive_directories():
    logger.debug("Testing if the container is running in user namespace.")
    notes = "No Docker containers found or docker is not running."

    results = GroupTestResult()

    containers = _get_docker_container()

    testcmd = '{{ .Id }}: Volumes={{ .Volumes }} VolumesRW={{ .VolumesRW }}'

    if not containers:
        return TestResult(Result.SKIP, notes)

    for container_id in containers:
        if container_id == '':
            pass
        else:
            check = "Checking container: " + str(container_id)
            test = subprocess.check_output(['docker',
                                            'inspect',
                                            '--format',
                                            testcmd,
                                            container_id])

            if ':true' in test:
                notes = ("Container " + str(container_id) + " has "
                         "sensitive host system directories " +
                         "mounted.")
                result = TestResult(Result.FAIL, notes)
            else:
                result = TestResult(Result.PASS)

            results.add_result(check, result)
    return results

def test_restart_policy():
    logger.debug("Testing if the container is running in user namespace.")
    notes = "No Docker containers found or docker is not running."

    results = GroupTestResult()

    containers = _get_docker_container()

    testcmd = '''{{ .Id }}: RestartPolicyName={{ .HostConfig.RestartPolicy.Name }}
    MaximumRetryCount={{ .HostConfig.RestartPolicy.MaximumRetryCount }}'''

    if not containers:
        return TestResult(Result.SKIP, notes)

    for container_id in containers:
        if container_id == '':
            pass
        else:
            check = "Checking container: " + str(container_id)
            test = subprocess.check_output(['docker',
                                            'inspect',
                                            '--format',
                                            testcmd,
                                            container_id])
            try:
                entry = test.split(':')
                r = entry[1].split('=')
                restart_policy = r[1].split(" ")
                max_retry = r[2]
                policy = str(restart_policy[0])

            except IndexError:
                notes = ("Container: " + str(container_id) + "returns "
                         "a malformed restart policy value.")
                result = TestResult(Result.SKIP, notes)
            else:
                if 'no' in policy or policy == " ":
                    result = TestResult(Result.PASS)
                elif policy is None:
                    result = TestResult(Result.PASS)
                elif policy == 'always':
                    notes = ("Container " + str(container_id) + " will always "
                             "restart regardless of max retry count. This is "
                             " not recommended.")
                    result = TestResult(Result.FAIL, notes)
                elif policy == 'on-failure':
                    if int(max_retry) <= 5:
                        result = TestResult(Result.PASS)
                    else:
                        notes = ("Container " + str(container_id) + " max "
                                 "retry count set to a non-compliant level.")
                        result = TestResult(Result.FAIL, notes)
                else:
                    notes = ("Cannot test. Container " + str(container_id) +
                             " settings not returning an expected value.")
                    result = TestResult(Result.SKIP, notes)
                results.add_result(check, result)
    return results

def body_size(config):
    try:
        path = os.path.join(config['dir'], 'cinder.conf')
        cinder_conf = utils.parse_openstack_ini(path)
    except EnvironmentError:
        return TestResult(Result.SKIP, 'cannot read cinder config files')

    osapi_max_body_size = int(cinder_conf.get('DEFAULT', {}).get(
        'osapi_max_request_body_size', '114688'))
    oslo_max_body_size = int(cinder_conf.get('oslo_middleware', {}).get(
        'max_request_body_size', '114688'))

    results = GroupTestResult()

    res_name = 'osapi body size'
    if osapi_max_body_size <= 114688:
        results.add_result(res_name, TestResult(Result.PASS))
    else:
        results.add_result(res_name, TestResult(
            Result.FAIL, 'osapi allows too big request bodies'))

    res_name = 'oslo body size'
    if oslo_max_body_size <= 114688:
        results.add_result(res_name, TestResult(Result.PASS))
    else:
        results.add_result(res_name, TestResult(
            Result.FAIL, 'middleware allows too big request bodies'))

    return results

def nas_security(config):
    try:
        path = os.path.join(config['dir'], 'cinder.conf')
        cinder_conf = utils.parse_openstack_ini(path)
    except EnvironmentError:
        return TestResult(Result.SKIP, 'cannot read cinder config files')

    secure_operations = cinder_conf.get('DEFAULT', {}).get(
        'nas_secure_file_operations', 'auto').lower() != 'false'
    secure_permissions = cinder_conf.get('DEFAULT', {}).get(
        'nas_secure_file_permissions', 'auto').lower() != 'false'

    results = GroupTestResult()

    if secure_operations:
        results.add_result('operations', TestResult(Result.PASS))
    else:
        results.add_result('operations', TestResult(
            Result.FAIL, 'NAS operations are not secure'))

    if secure_permissions:
        results.add_result('permissions', TestResult(Result.PASS))
    else:
        results.add_result('permissions', TestResult(
            Result.FAIL, 'NAS permissions are not secure'))

    return results

def safe_config(expected_config):
    if not os.path.exists(CONFIG_PATH):
        return TestResult(Result.SKIP, "MySQL config not found")

    try:
        config_lines = _get_full_config(CONFIG_PATH)
    except IOError:
        return TestResult(Result.FAIL, "MySQL config could not be read")
    results = GroupTestResult()
    for test, res in utils.verify_config(
            CONFIG_PATH, config_lines, expected_config, keyval_delim='='):
        results.add_result(test, res)
    return results

def no_exec(noexec_mounts):
    mounts = _get_mounts()

    results = GroupTestResult()
    for destination in noexec_mounts:
        point = _find_mount_point(mounts, destination.encode('utf-8'))

        if b'noexec' in point[3]:
            results.add_result(destination, TestResult(Result.PASS))
        else:
            dest = point[1].decode('utf-8', errors='replace')
            msg = "executable files allowed on %s" % (dest,)
            results.add_result(destination, TestResult(Result.FAIL, msg))
    return results