Test Failed
Push — master ( d68f82...e7d5ba )
by Jan
02:29 queued 12s
created

ssg.products   A

Complexity

Total Complexity 26

Size/Duplication

Total Lines 152
Duplicated Lines 0 %

Test Coverage

Coverage 84.15%

Importance

Changes 0
Metric Value
eloc 100
dl 0
loc 152
ccs 69
cts 82
cp 0.8415
rs 10
c 0
b 0
f 0
wmc 26

7 Functions

Rating   Name   Duplication   Size   Complexity  
A _validate_product_oval_feed_url() 0 10 3
A get_all() 0 24 3
A load_product_yaml() 0 27 1
A product_yaml_path() 0 2 1
F _get_implied_properties() 0 38 14
A get_profiles_directory() 0 5 2
A get_profile_files_from_root() 0 8 2
1 2
from __future__ import absolute_import
2 2
from __future__ import print_function
3
4 2
import os
5 2
from collections import namedtuple
6 2
from glob import glob
7
8 2
from .build_cpe import ProductCPEs
9 2
from .constants import (product_directories,
10
                        DEFAULT_GID_MIN,
11
                        DEFAULT_UID_MIN,
12
                        DEFAULT_GRUB2_BOOT_PATH,
13
                        DEFAULT_GRUB2_UEFI_BOOT_PATH,
14
                        DEFAULT_DCONF_GDM_DIR,
15
                        DEFAULT_AIDE_CONF_PATH,
16
                        DEFAULT_AIDE_BIN_PATH,
17
                        DEFAULT_SSH_DISTRIBUTED_CONFIG,
18
                        PKG_MANAGER_TO_SYSTEM,
19
                        PKG_MANAGER_TO_CONFIG_FILE,
20
                        XCCDF_PLATFORM_TO_PACKAGE,
21
                        SSG_REF_URIS)
22 2
from .utils import merge_dicts, required_key
23 2
from .yaml import open_raw
24
25
26 2
def _validate_product_oval_feed_url(contents):
27 2
    if "oval_feed_url" not in contents:
28 2
        return
29 2
    url = contents["oval_feed_url"]
30 2
    if not url.startswith("https"):
31
        msg = (
32
            "OVAL feed of product '{product}' is not available through an encrypted channel: {url}"
33
            .format(product=contents["product"], url=url)
34
        )
35
        raise ValueError(msg)
36
37
38 2
def _get_implied_properties(existing_properties):
39 2
    result = existing_properties.copy()
40 2
    if "pkg_manager" in existing_properties:
41 2
        pkg_manager = existing_properties["pkg_manager"]
42 2
        if "pkg_system" not in existing_properties:
43 2
            result["pkg_system"] = PKG_MANAGER_TO_SYSTEM[pkg_manager]
44 2
        if "pkg_manager_config_file" not in existing_properties:
45 2
            if pkg_manager in PKG_MANAGER_TO_CONFIG_FILE:
46 2
                result["pkg_manager_config_file"] = PKG_MANAGER_TO_CONFIG_FILE[pkg_manager]
47
48 2
    if "gid_min" not in existing_properties:
49 2
        result["gid_min"] = DEFAULT_GID_MIN
50
51 2
    if "uid_min" not in existing_properties:
52 2
        result["uid_min"] = DEFAULT_UID_MIN
53
54 2
    if "auid" not in existing_properties:
55 2
        result["auid"] = existing_properties.get("uid_min", DEFAULT_UID_MIN)
56
57 2
    if "grub2_boot_path" not in existing_properties:
58 2
        result["grub2_boot_path"] = DEFAULT_GRUB2_BOOT_PATH
59
60 2
    if "grub2_uefi_boot_path" not in existing_properties:
61 2
        result["grub2_uefi_boot_path"] = DEFAULT_GRUB2_UEFI_BOOT_PATH
62
63 2
    if "dconf_gdm_dir" not in existing_properties:
64 2
        result["dconf_gdm_dir"] = DEFAULT_DCONF_GDM_DIR
65
66 2
    if "aide_conf_path" not in existing_properties:
67 2
        result["aide_conf_path"] = DEFAULT_AIDE_CONF_PATH
68
69 2
    if "aide_bin_path" not in existing_properties:
70 2
        result["aide_bin_path"] = DEFAULT_AIDE_BIN_PATH
71
72 2
    if "sshd_distributed_config" not in existing_properties:
73 2
        result["sshd_distributed_config"] = DEFAULT_SSH_DISTRIBUTED_CONFIG
74
75 2
    return result
76
77
78 2
def product_yaml_path(ssg_root, product):
79
    return os.path.join(ssg_root, "products", product, "product.yml")
80
81
82 2
def load_product_yaml(product_yaml_path):
83
    """
84
    Reads a product data from disk and returns it.
85
    The returned product dictionary also contains derived useful information.
86
    """
87
88 2
    product_yaml = open_raw(product_yaml_path)
89 2
    _validate_product_oval_feed_url(product_yaml)
90
91
    # The product directory is necessary to get absolute paths to benchmark, profile and
92
    # cpe directories, which are all relative to the product directory
93 2
    product_yaml["product_dir"] = os.path.dirname(product_yaml_path)
94
95 2
    platform_package_overrides = product_yaml.get("platform_package_overrides", {})
96
    # Merge common platform package mappings, while keeping product specific mappings
97 2
    product_yaml["platform_package_overrides"] = merge_dicts(XCCDF_PLATFORM_TO_PACKAGE,
98
                                                             platform_package_overrides)
99 2
    product_yaml.update(_get_implied_properties(product_yaml))
100
101
    # The product_yaml should be aware of the ProductCPEs
102 2
    product_yaml["product_cpes"] = ProductCPEs(product_yaml)
103
104 2
    reference_uris = product_yaml.get("reference_uris", {})
105 2
    product_yaml["reference_uris"] = merge_dicts(SSG_REF_URIS,
106
                                                 reference_uris)
107
108 2
    return product_yaml
109
110
111 2
def get_all(ssg_root):
112
    """
113
    Analyzes all products in the SSG root and sorts them into two categories:
114
    those which use linux_os and those which use their own directory. Returns
115
    a namedtuple of sets, (linux, other).
116
    """
117
118 2
    linux_products = set()
119 2
    other_products = set()
120
121 2
    for product in product_directories:
122 2
        product_yaml_path = os.path.join(ssg_root, "products", product, "product.yml")
123 2
        product_yaml = load_product_yaml(product_yaml_path)
124
125 2
        guide_dir = os.path.join(product_yaml["product_dir"], product_yaml['benchmark_root'])
126 2
        guide_dir = os.path.abspath(guide_dir)
127
128 2
        if 'linux_os' in guide_dir:
129 2
            linux_products.add(product)
130
        else:
131 2
            other_products.add(product)
132
133 2
    products = namedtuple('products', ['linux', 'other'])
134 2
    return products(linux_products, other_products)
135
136
137 2
def get_profiles_directory(env_yaml):
138
    profiles_root = None
139
    if env_yaml:
140
        profiles_root = required_key(env_yaml, "profiles_root")
141
    return profiles_root
142
143
144 2
def get_profile_files_from_root(env_yaml, product_yaml):
145
    profile_files = []
146
    if env_yaml:
147
        profiles_root = get_profiles_directory(env_yaml)
148
        base_dir = os.path.dirname(product_yaml)
149
        profile_files = sorted(glob("{base_dir}/{profiles_root}/*.profile"
150
                               .format(profiles_root=profiles_root, base_dir=base_dir)))
151
    return profile_files
152